Intel E810/ICE DPU RDMA 及MLX中断原理分析1(CE/AE)

用户态确认完成事件流程:
ibv_ack_cq_events(cb->cq, 1)
	pthread_mutex_lock(&cq->mutex);
    cq->comp_events_completed += nevents
    pthread_cond_signal(&cq->cond) -> 唤醒其他线程(如:销毁CQ) -> pthread_cond_wait(&cq->cond, &cq->mutex) <- ibv_cmd_destroy_cq -> 在销毁CQ中等待所有完成事件和异步事件完成
    pthread_mutex_unlock(&cq->mutex)

在RDMA（远程直接内存访问）上下文中，**请求完成（SC）** 是一种特定类型的完成，与操作相关联，其中 RDMA 硬件在操作完成时明确通知启动器，但仅在请求时（即请求完成）。

关键概念：
完成队列（CQ）：

使用完成队列 (CQ)跟踪 RDMA 操作。当 RDMA 操作（例如，读取、写入、发送、接收）完成时，将在关联的 CQ 中发布完成事件，通知应用程序操作已完成。
请求完成与非请求完成：

请求完成 (SC)：这是由 RDMA 操作发起者明确请求的完成。只有在明确请求时才会将完成发布到 CQ，通常是通过 RDMA 操作中的标志或特定设置。
未经请求的完成 (UC)：这是每当某个操作完成时由 RDMA 设备自动发布的完成，无论应用程序是否明确请求它。
实践中请求完成：

在许多 RDMA 实现中（例如使用 InfiniBand或RoCE协议的实现），请求完成用于以下上下文中：
RDMA 写入或发送：发起者可能不关心完成事件，除非它明确请求完成事件。当应用程序只对某些完成感兴趣时，这可以减少开销并提高效率。
内存区域访问：对于内存读取或写入，可以请求需要确认的操作完成，但前提是用户指定。
例如，在RDMA 写操作中（尤其是IBV_WR_RDMA_WRITE），可以根据标志来请求完成，这些标志告诉硬件在操作完成时通知完成队列。
如何使用征求完成：

当发出 RDMA 操作时，应用程序可以在工作请求 (WR) 中设置一个Solicited Completion Flag。此标志确保只有当操作明确完成时才会将完成发布到 CQ，从而允许应用程序控制何时接收完成通知。
请求完成的主要优点是它们允许应用程序限制其需要处理的完成通知的数量，从而减少开销。
请求完成标志的示例：
在 RDMA 操作（例如 RDMA 发送或写入）中发布工作请求 (WR) 时，应用程序可以指定完成是经过请求的还是未经请求的。这可以使用特定标志来完成。

对于带有库的InfiniBand或RoCElibibverbs的情况，例如：

IBV_SEND_SOLICITED：用于指示应请求此发送操作完成的标志。
下面是代码中可能出现的一个基本示例：

struct ibv_send_wr wr = {
    .wr_id = 1,                       // Unique ID for the work request
    .sg_list = &sg,                    // Scatter-gather list
    .num_sge = 1,                      // Number of scatter-gather entries
    .opcode = IBV_WR_SEND,             // Type of work request (send)
    .send_flags = IBV_SEND_SOLICITED,  // Request solicited completion
    .next = NULL
};

// Post the work request
if (ibv_post_send(qp, &wr, &bad_wr)) {
    // Handle error
}
要点：
减少完成队列噪音：通过使用请求的完成，您可以避免在完成队列中出现过多通知，这在您执行应用程序仅对某些完成事件感兴趣的操作时很有用。
效率：对于 RDMA 写入之类的操作，应用程序可能不需要跟踪每个单独操作的完成情况，请求的完成可以帮助避免不必要的工作并提高总体吞吐量。
使用案例：
高性能计算 (HPC)：执行许多 RDMA 操作（例如批量数据传输）并且只需要完成某些关键操作时。
低延迟应用程序：在交易系统、数据库或其他低延迟系统中，有效管理完成通知至关重要。
概括：
RDMA 中的请求完成 (SC)是仅在应用程序明确请求时才发布到完成队列的完成事件。此功能有助于减少不必要的通知、提高性能，并更好地控制应用程序何时接收完成事件。

以太探测函数:
ice_probe
err = ice_alloc_vsis(pf)
pf->vsi = devm_kcalloc(dev, pf->num_alloc_vsi, sizeof(*pf->vsi)
err = ice_init_eth(pf)
ice_for_each_q_vector(vsi, v_idx)
netif_napi_add(vsi->netdev, &vsi->q_vectors[v_idx]->napi, ice_napi_poll) -> NAPI polling Rx/Tx cleanup routine
ret = ice_alloc_rdma_qvectors(pf) /* Reserve vector resources */ 预留中断资源, 在下面的函数详情中获取中断号(IRQ)并通过数组管理

...
在设备初始化中设置PF上的RDMA中断数量
static int ice_init_dev(struct ice_pf *pf)
	err = ice_init_interrupt_scheme(pf) -> 确定适当的中断方案, @pf：要初始化的板级私有结构
   		vectors = ice_ena_msix_range(pf) -> 启用中断向量范围
    		pf->num_rdma_msix = num_cpus + ICE_RDMA_NUM_AEQ_MSIX -> CPU核数 + 4个AEQ中断 -> 设置中断数量
    		v_actual = pci_alloc_irq_vectors(pf->pdev, ICE_MIN_MSIX, v_wanted, PCI_IRQ_MSIX) -> 分配中断向量,得到实际分配的中断向量数

RDMA探测函数:
static int irdma_probe
struct ice_pf *pf = iidc_adev->pf
irdma_fill_device_info(iwdev, pf, vsi) -> 填充设备能力
    rf->hw.hw_addr = pf->hw.hw_addr; <- info.bar0 = rf->hw.hw_addr;
    rf->msix_count =  pf->num_rdma_msix -> 设置RDMA中断数量到 msix_count 属性上
    
// 中断MAP
struct msi_map {
	int	index;
	int	virq;
};

// RDMA业务的中断对象
struct irdma_msix_vector {
	u32 idx;
	u32 irq;
	u32 cpu_affinity;
	u32 ceq_id;
	cpumask_t mask;
	char name[IRDMA_IRQ_NAME_STR_LEN];
};


// 队列的中断信息
struct irdma_qvlist_info {
	u32 num_vectors;
	struct irdma_qv_info qv_info[];
};

// 队列的中断信息-详情    
struct irdma_qv_info {
	u32 v_idx; /* msix_vector */
	u16 ceq_idx;
	u16 aeq_idx;
	u8 itr_idx;
};

err = ice_init_rdma(pf)
    ice_is_rdma_ena(pf)
        return test_bit(ICE_FLAG_RDMA_ENA, pf->flags) -> Check RDMA ENABLE/DISABLE(检查是否启用RDMA功能)
    ret = ice_alloc_rdma_qvectors(pf) -> ice：添加单独的中断分配，目前中断分配，根据某个特性是批量分配的。 此外，分配后还有一系列操作，通过该批中断分配每个 irq 设置。 尽管驱动程序尚不支持动态中断分配，但将分配的中断保留在池中并添加分配抽象逻辑以使代码更加灵活。 将每个中断信息保留在 ice_q_vector 结构中，这会产生ice_vsi::base_vector冗余。 此外，因此有一些功能可以删除
        pf->msix_entries = kcalloc(pf->num_rdma_msix, sizeof(*pf->msix_entries), GFP_KERNEL) -> 分配管理中断的数组
        for (i = 0; i < pf->num_rdma_msix; i++)
            struct msix_entry *entry = &pf->msix_entries[i]
            struct msi_map map
            map = ice_alloc_irq(pf, false) -> ice：添加动态中断分配，目前驱动程序只能在init阶段通过调用pci_alloc_irq_vectors分配中断向量。 对此进行更改并使用新的 pci_msix_alloc_irq_at/pci_msix_free_irq API，并在启用 MSIX 后启用分配和释放更多中断。 由于并非所有平台都支持动态分配，请使用 pci_msix_can_alloc_dyn 检查。 扩展跟踪器以跟踪最初分配的中断数量，因此当所有此类向量都已使用时，会自动动态分配其他中断。 记住每个中断分配方法，然后适当地释放。 由于某些功能可能需要动态分配的中断，因此添加适当的 VSI 标志并在分配新中断时将其考虑在内
            为给定所有者 ID 分配新的中断向量。 返回包含中断详细信息的 struct msi_map 并适当跟踪分配的中断。 该函数从 irq_tracker 保留新的 irq 条目。 如果根据跟踪器信息，使用ice_pci_alloc_irq_vectors分配的所有中断都已使用并且支持动态分配的中断，则将使用pci_msix_alloc_irq_at分配新中断。 一些调用者可能只支持动态分配的中断。 这由 dyn_only 标志指示。 失败时，返回 .index 为负的映射。 调用者应该检查返回的map索引
                struct ice_irq_entry *entry
                entry = ice_get_irq_res(pf, dyn_only)
                    entry = kzalloc(sizeof(*entry), GFP_KERNEL)
                    ret = xa_alloc(&pf->irq_tracker.entries, &index, entry, limit, GFP_KERNEL) -> 跟踪中断
                    entry->index = index
                    entry->dynamic = index >= num_static
                pci_msix_can_alloc_dyn
                pci_msix_alloc_irq_at or
                map.index = entry->index
                map.virq = pci_irq_vector(pf->pdev, map.index) -> 获取中断号并纳入管理
            entry->entry = map.index;
            entry->vector = map.virq

irdma_ctrl_init_hw -> 初始化硬件的控制部分
    irdma_setup_init_state
        status = irdma_save_msix_info(rf) -> 保存中断信息 -> 将 msix 矢量信息复制到 iwarp 设备 @rf: RDMA PCI 函数分配, iwdev msix 表并将 msix 信息复制到表中如果成功则返回 0，否则返回错误
            rf->iw_msixtbl = kzalloc(size, GFP_KERNEL)
            pmsix = rf->msix_entries
            for (i = 0, ceq_idx = 0; i < rf->msix_count; i++, iw_qvinfo++)
                rf->iw_msixtbl[i].idx = pmsix->entry;
                rf->iw_msixtbl[i].irq = pmsix->vector;
                rf->iw_msixtbl[i].cpu_affinity = ceq_idx;                

irdma_setup_ceq_0 -> 创建CEQ 0及其中断资源 -> 为所有设备完成事件队列分配一个列表，创建ceq 0并配置其msix中断向量返回0，如果设置成功，否则返回错误
    num_ceqs = min(rf->msix_count, rf->sc_dev.hmc_fpm_misc.max_ceqs
    rf->ceqlist = kcalloc(num_ceqs, sizeof(*rf->ceqlist), GFP_KERNEL)
    irdma_create_ceq(rf, iwceq, 0, &rf->default_vsi) -> 创建完成事件队列
        irdma_sc_ceq_init(&iwceq->sc_ceq, &info)
        irdma_cqp_ceq_cmd(&rf->sc_dev, &iwceq->sc_ceq, IRDMA_OP_CEQ_CREATE)
        or irdma_sc_cceq_create(&iwceq->sc_ceq, 0)
    i = rf->msix_shared ? 0 : 1 -> 共享模式取第0个索引
    irdma_cfg_ceq_vector(rf, iwceq, 0, msix_vec) -> 为完成事件队列设置中断
        if (rf->msix_shared && !ceq_id)
            tasklet_setup(&rf->dpc_tasklet, irdma_dpc)
            request_irq(msix_vec->irq, irdma_irq_handler, 0, -> 注册中断服务(为中断线注册中断控制器), 回调函数为: irdma_dpc
        else
            tasklet_setup(&iwceq->dpc_tasklet, irdma_ceq_dpc)
                irdma_process_ceq(rf, iwceq)
                    do cq = irdma_sc_process_ceq(dev, sc_ceq)
                        ceqe = IRDMA_GET_CURRENT_CEQ_ELEM(ceq)
                        ceq->polarity ^= 1
                        irdma_sc_cq_ack(cq)
                    queue_work(rf->cqp_cmpl_wq, &rf->cqp_cmpl_work)
                    irdma_puda_ce_handler(rf, cq)
                        do {
                            irdma_puda_poll_cmpl(dev, cq, &compl_error)
                                if (info.q_type == IRDMA_CQE_QTYPE_RQ)
                                    irdma_puda_poll_info(cq, &info)
                                        cqe = IRDMA_GET_CURRENT_CQ_ELEM(&cq->cq_uk)
                                            (_cq)->cq_base[IRDMA_RING_CURRENT_HEAD((_cq)->cq_ring)].buf
                                        get_64bit_val(cqe, 24, &qword3)
                                        info->qp = (struct irdma_qp_uk *)(unsigned long)comp_ctx
                                        ...
                                    dma_sync_single_for_cpu -> 确保DMA缓冲区中的数据与物理内存中的数据同步。如果需要将数据从设备上读取到内存中，则应该使用 dma_sync_single_for_cpu()函数。如果需要将数据从内存中写入到设备上，则应该使用 dma_sync_single_for_device()函数
                                    irdma_puda_get_tcpip_info -> get tcpip info from puda buffer
                                        if (buf->vsi->dev->hw_attrs.uk_attrs.hw_rev == IRDMA_GEN_1)
                                            irdma_gen1_puda_get_tcpip_info(info, buf)
                                                if (ethh->h_proto == htons(0x8100))
                                                    buf->vlan_id = ntohs(((struct vlan_ethhdr *)ethh)->h_vlan_TCI) & VLAN_VID_MASK
                                                    ip6h = (struct ipv6hdr *)buf->iph
                                        buf->ipv4 = info->ipv4
                                        buf->seqnum = ntohl(tcph->seq)
                                        ether_addr_copy(buf->smac, info->smac)
                                    rsrc->receive(rsrc->vsi, buf) -> irdma_ieq_receive
                                        qp = irdma_ieq_get_qp(vsi->dev, buf)
                                        irdma_ieq_handle_exception(ieq, qp, buf)
                                            irdma_ieq_check_first_buf(buf, fps)
                                            irdma_send_ieq_ack(qp)
                                    irdma_ilq_putback_rcvbuf
                                    or irdma_puda_replenish_rq
                                        irdma_puda_get_bufpool
                                            irdma_puda_get_listbuf
                                        irdma_puda_post_recvbuf
                                else
                                    rsrc->xmit_complete(rsrc->vsi, buf -> irdma_ieq_tx_compl
                                        irdma_puda_ret_bufpool
                                            list_add(&buf->list, &rsrc->bufpool)
                                    irdma_puda_send_buf
                                        irdma_puda_send
                                            wqe = irdma_puda_get_next_send_wqe(&qp->qp_uk, &wqe_idx)
                                            irdma_uk_qp_post_wr(&qp->qp_uk)
                                                writel(qp->qp_id, qp->wqe_alloc_db)
                            irdma_sc_ccq_arm(cq)
                                writel(ccq->cq_uk.cq_id, ccq->dev->cq_arm_db)
                        }
                irdma_ena_intr(&rf->sc_dev, iwceq->msix_idx) -> dev->irq_ops->irdma_en_irq(dev, msix_id) -> icrdma_ena_irq -> 启用中断
                    writel(val, dev->hw_regs[IRDMA_GLINT_DYN_CTL] + idx)
            status = request_irq(msix_vec->irq, irdma_ceq_handler, 0, msix_vec->name, iwceq) -> 注册中断服务(为中断线注册中断控制器)
                tasklet_schedule(&iwceq->dpc_tasklet)  -> 回调函数为: irdma_ceq_dpc

static int create_cq
    ...
   	cq->comp_handler  = ib_uverbs_comp_handler;
	cq->event_handler = ib_uverbs_cq_event_handler;
	...

static int UVERBS_HANDLER(UVERBS_METHOD_CQ_CREATE)
	...
	 cq->comp_handler  = ib_uverbs_comp_handler;

	 cq->event_handler = ib_uverbs_cq_event_handler;
	...

ibv_create_comp_channel
IB_USER_VERBS_CMD_CREATE_COMP_CHANNEL -> ib_uverbs_create_comp_channel
    struct ib_uverbs_completion_event_file	  *ev_file
    ib_uverbs_init_event_queue(&ev_file->ev_queue)
        spin_lock_init(&ev_queue->lock)
        INIT_LIST_HEAD(&ev_queue->event_list);
        init_waitqueue_head(&ev_queue->poll_wait);
        ev_queue->is_closed   = 0;
        ev_queue->async_queue = NULL;
    resp.fd = uobj->id -> 关联文件描述符

CQ中断模式:
ibv_req_notify_cq(pp_cq(ctx), 0) -> 请求 CQ 上的完成通知。当条目添加到 CQ 时，将向与 CQ 关联的完成通道添加事件。@cq：请求通知的完成队列。@solicited_only：如果非零，则仅为下一个请求的 CQ 条目生成事件(SE模式)。如果为零，则任何 CQ 条目（无论是否请求）都将生成事件
    cq->context->ops.req_notify_cq(cq, solicited_only)
    
.req_notify_cq = irdma_uarm_cq,
    enum irdma_cmpl_notify cq_notify = IRDMA_CQ_COMPL_EVENT
    irdma_arm_cq(iwucq, cq_notify)
        iwucq->is_armed = true;
        iwucq->arm_sol = true;
        iwucq->skip_arm = false;
        iwucq->skip_sol = true;
        irdma_uk_cq_request_notification(&iwucq->cq, cq_notify)
            get_64bit_val(cq->shadow_area, 32, &temp_val) -> 从CQ上获取shadow_area到temp_val
            arm_seq_num = (__u8)FIELD_GET(IRDMA_CQ_DBSA_ARM_SEQ_NUM, temp_val);
            arm_seq_num++;
            if (cq_notify == IRDMA_CQ_COMPL_EVENT)
                arm_next = 1;
            temp_val =  FIELD_PREP(IRDMA_CQ_DBSA_ARM_SEQ_NUM, arm_seq_num) |
                        FIELD_PREP(IRDMA_CQ_DBSA_SW_CQ_SELECT, sw_cq_sel) |
                        FIELD_PREP(IRDMA_CQ_DBSA_ARM_NEXT_SE, arm_next_se) |
                        FIELD_PREP(IRDMA_CQ_DBSA_ARM_NEXT, arm_next); 
            set_64bit_val(cq->shadow_area, 32, temp_val);-> 更新shadow_area
            udma_to_device_barrier(); /* make sure WQE is populated before valid bit is set */
            db_wr32(cq->cq_id, cq->cqe_alloc_db)

ibv_get_cq_event(ctx->channel, &ev_cq, &ev_ctx)
    struct ib_uverbs_comp_event_desc ev
    if (read(channel->fd, &ev, sizeof ev) != sizeof ev)
    *cq         = (struct ibv_cq *) (uintptr_t) ev.cq_handle;
    *cq_context = (*cq)->cq_context;
    get_ops((*cq)->context)->cq_event(*cq); -> 执行.cq_event回调

E810实现:
void irdma_cq_event(struct ibv_cq *cq)
{
	struct irdma_ucq *iwucq;

	iwucq = container_of(cq, struct irdma_ucq, verbs_cq.cq);
	if (pthread_spin_lock(&iwucq->lock))
		return;

	if (iwucq->skip_arm)
		irdma_arm_cq(iwucq, IRDMA_CQ_COMPL_EVENT);
	else
		iwucq->is_armed = false;

	pthread_spin_unlock(&iwucq->lock);
}

MLX5实现:
void mlx5_cq_event(struct ibv_cq *cq)
{
  to_mcq(cq)->arm_sn++;
}

海思实现:
void hns_roce_u_cq_event(struct ibv_cq *cq)
{
	to_hr_cq(cq)->arm_sn++;
}

irdma_cfg_ceq_vector
    irdma_dpc
        irdma_process_aeq
            ...
            case IRDMA_AE_CQ_OPERATION_ERROR
                if (iwcq->ibcq.event_handler)
                    iwcq->ibcq.event_handler(&ibevent, iwcq->ibcq.cq_context) -> cq->event_handler = ib_uverbs_cq_event_handler; -> 处理CQ异步错误事件
                else
                    tasklet_setup(&iwceq->dpc_tasklet, irdma_ceq_dpc)
                        irdma_process_ceq(rf, iwceq)
                            do cq = irdma_sc_process_ceq(dev, sc_ceq)
                                ceq->polarity ^= 1
                                irdma_sc_cq_ack(cq)
                            if (cq->cq_type == IRDMA_CQ_TYPE_IWARP)
                                irdma_iwarp_ce_handler(cq)
                                    if (cq->ibcq.comp_handler)
                                        cq->ibcq.comp_handler(&cq->ibcq, cq->ibcq.cq_context) -> ib_uverbs_comp_handler -> 处理CQ异步完成事件

cq->comp_handler  = ib_uverbs_comp_handler
	list_add_tail(&entry->list, &ev_queue->event_list)
	list_add_tail(&entry->obj_list, &uobj->comp_list)
	wake_up_interruptible(&ev_queue->poll_wait)
	kill_fasync(&ev_queue->async_queue, SIGIO, POLL_IN) -> 唤醒正在等待事件的用户态(ibv_get_cq_event)

ib_uverbs_cq_event_handler
    uverbs_uobj_event(&event->element.cq->uobject->uevent, event)
        ib_uverbs_async_handler(eobj->event_file, eobj->uobject.user_handle, event->event, &eobj->event_list, &eobj->events_reported)
            list_add_tail(&entry->list, &async_file->ev_queue.event_list)
            list_add_tail(&entry->obj_list, obj_list)
            wake_up_interruptible(&async_file->ev_queue.poll_wait)
            kill_fasync(&async_file->ev_queue.async_queue, SIGIO, POLL_IN) -> 唤醒正在等待事件的用户态(ibv_get_cq_event)

mlx5_eq_table_create
    create_async_eqs
        MLX5_NB_INIT(&table->cq_err_nb, cq_err_event_notifier, CQ_ERROR) -> new
        mlx5_eq_notifier_register(dev, &table->cq_err_nb)

cq_err_event_notifier
	if (cq->event)
			cq->event(cq, type) -> mlx5_ib_cq_event
                if (ibcq->event_handler)
                    ibcq->event_handler(&event, ibcq->cq_context) -> ib_uverbs_cq_event_handler

int mlx5_arm_cq(struct ibv_cq *ibvcq, int solicited)
{
	struct mlx5_cq *cq = to_mcq(ibvcq);
	struct mlx5_context *ctx = to_mctx(ibvcq->context);
	uint64_t doorbell;
	uint32_t sn;
	uint32_t ci;
	uint32_t cmd;

	sn  = cq->arm_sn & 3;
	ci  = cq->cons_index & 0xffffff;
	cmd = solicited ? MLX5_CQ_DB_REQ_NOT_SOL : MLX5_CQ_DB_REQ_NOT;

	doorbell = sn << 28 | cmd | ci;
	doorbell <<= 32;
	doorbell |= cq->cqn;

	cq->dbrec[MLX5_CQ_ARM_DB] = htobe32(sn << 28 | cmd | ci);

	/*
	 * Make sure that the doorbell record in host memory is
	 * written before ringing the doorbell via PCI WC MMIO.
	 */
	mmio_wc_start();

	mmio_write64_be(ctx->cq_uar_reg + MLX5_CQ_DOORBELL, htobe64(doorbell));

	mmio_flush_writes();

	return 0;
}

enum ib_cq_notify_flags {
	IB_CQ_SOLICITED			= 1 << 0,
	IB_CQ_NEXT_COMP			= 1 << 1,
	IB_CQ_SOLICITED_MASK		= IB_CQ_SOLICITED | IB_CQ_NEXT_COMP,
	IB_CQ_REPORT_MISSED_EVENTS	= 1 << 2,
};

static int nvme_rdma_alloc_queue(struct nvme_rdma_ctrl *ctrl,
		int idx, size_t queue_size)
{
	struct nvme_rdma_queue *queue;
	struct sockaddr *src_addr = NULL;
	int ret;

	queue = &ctrl->queues[idx];
	mutex_init(&queue->queue_lock);
	queue->ctrl = ctrl;
	if (idx && ctrl->ctrl.max_integrity_segments)
		queue->pi_support = true;
	else
		queue->pi_support = false;
	init_completion(&queue->cm_done);

	if (idx > 0)
		queue->cmnd_capsule_len = ctrl->ctrl.ioccsz * 16;
	else
		queue->cmnd_capsule_len = sizeof(struct nvme_command);

	queue->queue_size = queue_size;

	queue->cm_id = rdma_create_id(&init_net, nvme_rdma_cm_handler, queue,
			RDMA_PS_TCP, IB_QPT_RC); // 创建RDMA_ID时设置CM回调
	if (IS_ERR(queue->cm_id)) {
		dev_info(ctrl->ctrl.device,
			"failed to create CM ID: %ld\n", PTR_ERR(queue->cm_id));
		ret = PTR_ERR(queue->cm_id);
		goto out_destroy_mutex;
	}

	if (ctrl->ctrl.opts->mask & NVMF_OPT_HOST_TRADDR)
		src_addr = (struct sockaddr *)&ctrl->src_addr;

	queue->cm_error = -ETIMEDOUT;
	ret = rdma_resolve_addr(queue->cm_id, src_addr,
			(struct sockaddr *)&ctrl->addr,
			NVME_RDMA_CM_TIMEOUT_MS);
	if (ret) {
		dev_info(ctrl->ctrl.device,
			"rdma_resolve_addr failed (%d).\n", ret);
		goto out_destroy_cm_id;
	}

	ret = nvme_rdma_wait_for_cm(queue);
	if (ret) {
		dev_info(ctrl->ctrl.device,
			"rdma connection establishment failed (%d)\n", ret);
		goto out_destroy_cm_id;
	}

	set_bit(NVME_RDMA_Q_ALLOCATED, &queue->flags);

	return 0;

out_destroy_cm_id:
	rdma_destroy_id(queue->cm_id);
	nvme_rdma_destroy_queue_ib(queue);
out_destroy_mutex:
	mutex_destroy(&queue->queue_lock);
	return ret;
}

nvme_rdma_cm_handler
    case RDMA_CM_EVENT_ADDR_RESOLVED:
            cm_error = nvme_rdma_addr_resolved(queue)
                ret = nvme_rdma_create_queue_ib(queue)
                    ret = nvme_rdma_create_cq(ibdev, queue)
                        queue->ib_cq = ib_cq_pool_get(ibdev, queue->cq_size, comp_vector, IB_POLL_SOFTIRQ) -> ib_cq_pool_get() - 查找与给定的 cpu 提示匹配（或对于通配符亲和性最少使用）且适合 nr_cqe 的最少使用的完成队列。@dev：rdma 设备 @nr_cqe：所需的 cqe 条目数 @comp_vector_hint：驱动程序根据内部计数器分配 comp 向量的完成向量提示（-1）@poll_ctx：cq 轮询上下文查找满足 @comp_vector_hint 和 @nr_cqe 要求的 cq 并为我们声明其中的条目。如果没有可用的 cq，则分配一个符合要求的新 cq 并将其添加到设备池中。IB_POLL_DIRECT 不能用于共享 cq，因此它不是 @poll_ctx 的有效值
                            ret = ib_alloc_cqs(dev, nr_cqe, poll_ctx) -> ib_alloc_cq
...
__ib_alloc_cq(dev, private, nr_cqe, comp_vector, poll_ctx, caller)
    cq = rdma_zalloc_drv_obj(dev, ib_cq)
    cq->comp_vector = comp_vector
    cq->wc = kmalloc_array(IB_POLL_BATCH, sizeof(*cq->wc), GFP_KERNEL) -> 16
    ret = dev->ops.create_cq(cq, &cq_attr, NULL) -> 
    rdma_dim_init
    switch (cq->poll_ctx)
    case IB_POLL_DIRECT:
        cq->comp_handler = ib_cq_completion_direct
    case IB_POLL_SOFTIRQ // 软中断
        cq->comp_handler = ib_cq_completion_softirq
            irq_poll_sched(&cq->iop)
                list_add_tail(&iop->list, this_cpu_ptr(&blk_cpu_iopoll))
                raise_softirq_irqoff(IRQ_POLL_SOFTIRQ) // 唤醒
        irq_poll_init(&cq->iop, IB_POLL_BUDGET_IRQ, ib_poll_handler) -> IB：添加适当的完成队列抽象，这添加了一个抽象，允许 ULP 简单地向每
个提交的 WR 传递完成对象和完成回调，并让 RDMA 核心处理如何处理完成中断和轮询 CQ 的具体细节。   详细来说，有一个新的 ib_cqe 结构，它只
包含完成回调，并且可用于使用 container_of 获取包含对象。 WR 和 WC 指出它作为 wr_id 字段的替代项，类似于有多少 ULP 已经使用该字段通过
强制转换来存储指针。   使用新的完成回调的驱动程序使用新的 ib_create_cq API 分配它的 CQ，除了 CQE 的数量和完成向量之外，它还采用我们如
何轮询 CQE 的模式。 提供三种模式：直接用于从不接受 CQ 中断并仅轮询它们的驱动程序，软中断使用重命名的 blk-iopoll 基础设施从软中断上下文
进行轮询，该基础设施负责重新配置和预算，或者为想要的消费者提供工作队列 从用户上下文中调用。   非常感谢 Sagi Grimberg，他帮助审查了 
API，编写了当前版本的工作队列代码，因为我之前的两次尝试太糟糕了，并将 iSER 启动器转换为新的 API
            iop->poll = ib_poll_handler <- irq_poll_softirq -> poll()
                completed = __ib_process_cq(cq, budget, cq->wc, IB_POLL_BATCH)
                    trace_cq_process(cq)
                    while ((n = __poll_cq(cq, min_t(u32, batch,
                        ib_poll_cq(cq, num_entries, wc)
                            return cq->device->ops.poll_cq(cq, num_entries, wc)
						wc->wr_cqe->done(cq, wc)
                if (completed < budget)
                    irq_poll_complete(&cq->iop)
                    irq_poll_sched(&cq->iop)
                rdma_dim
        ib_req_notify_cq(cq, IB_CQ_NEXT_COMP) -> 内核请求完成中断通知
			return cq->device->ops.req_notify_cq(cq, flags)                                                    	

.req_notify_cq = irdma_req_notify_cq,
    cq_notify = notify_flags == IB_CQ_SOLICITED ? IRDMA_CQ_COMPL_SOLICITED : IRDMA_CQ_COMPL_EVENT
    Only promote to arm the CQ for any event if the last arm event was solicited
    if (!atomic_cmpxchg(&iwcq->armed, 0, 1) || promo_event)
        iwcq->last_notify = cq_notify
        irdma_uk_cq_request_notification(ukcq, cq_notify)
            get_64bit_val(cq->shadow_area, 32, &temp_val)
            arm_seq_num = (u8)FIELD_GET(IRDMA_CQ_DBSA_ARM_SEQ_NUM, temp_val)
            arm_seq_num++;
            if (cq_notify == IRDMA_CQ_COMPL_EVENT)
                arm_next = 1;
            set_64bit_val(cq->shadow_area, 32, temp_val)
            writel(cq->cq_id, cq->cqe_alloc_db)

static void irdma_process_aeq(struct irdma_pci_f *rf)
   	...
	case IRDMA_AE_CQ_OPERATION_ERROR:
			ibdev_err(&iwdev->ibdev,
				  "Processing an iWARP related AE for CQ misc = 0x%04X\n",
				  info->ae_id);

			spin_lock_irqsave(&rf->cqtable_lock, flags);
			iwcq = rf->cq_table[info->qp_cq_id];
			if (!iwcq) {
				spin_unlock_irqrestore(&rf->cqtable_lock,
						       flags);
				ibdev_dbg(to_ibdev(dev),
					  "cq_id %d is already freed\n", info->qp_cq_id);
				continue;
			}
			irdma_cq_add_ref(&iwcq->ibcq); // 加引用
			spin_unlock_irqrestore(&rf->cqtable_lock, flags);

			if (iwcq->ibcq.event_handler) {
				struct ib_event ibevent;

				ibevent.device = iwcq->ibcq.device;
				ibevent.event = IB_EVENT_CQ_ERR;
				ibevent.element.cq = &iwcq->ibcq;
				iwcq->ibcq.event_handler(&ibevent,
							 iwcq->ibcq.cq_context);
			}
			irdma_cq_rem_ref(&iwcq->ibcq); // 减引用
			break;
...

	.req_notify_cq = mlx5_ib_arm_cq,
		if (cq->notify_flags != IB_CQ_NEXT_COMP)
			cq->notify_flags = flags & IB_CQ_SOLICITED_MASK
		if ((flags & IB_CQ_REPORT_MISSED_EVENTS) && !list_empty(&cq->wc_list))
			ret = 1;
		mlx5_cq_arm(&cq->mcq, (flags & IB_CQ_SOLICITED_MASK) == IB_CQ_SOLICITED ? MLX5_CQ_DB_REQ_NOT_SOL : MLX5_CQ_DB_REQ_NOT, uar_page, to_mcq(ibcq)->mcq.cons_index)
			sn = cq->arm_sn & 3
			ci = cons_index & 0xffffff
			*cq->arm_db = cpu_to_be32(sn << 28 | cmd | ci)
			wmb()
			doorbell[0] = cpu_to_be32(sn << 28 | cmd | ci);
			doorbell[1] = cpu_to_be32(cq->cqn);
			mlx5_write64(doorbell, uar_page + MLX5_CQ_DOORBELL)

struct irdma_pci_f {
	bool reset:1;
	bool rsrc_created:1;
	bool msix_shared:1; // false
    ...

irdma_probe
	irdma_ctrl_init_hw
    	irdma_setup_init_state
    		irdma_save_msix_info
    			if (rf->msix_count <= num_online_cpus())
                    rf->msix_shared = true; // 如果CPU核数比需求的中断数小(cpu核不够多), 则启用中断共享
...
    
// 保存中断向量信息实现细节:    
static int irdma_save_msix_info(struct irdma_pci_f *rf)
{
	struct irdma_qvlist_info *iw_qvlist;
	struct irdma_qv_info *iw_qvinfo;
	struct msix_entry *pmsix;
	u32 ceq_idx;
	u32 i;
	size_t size;

	if (!rf->msix_count)
		return -EINVAL;

	size = sizeof(struct irdma_msix_vector) * rf->msix_count;
	size += struct_size(iw_qvlist, qv_info, rf->msix_count);
	rf->iw_msixtbl = kzalloc(size, GFP_KERNEL);
	if (!rf->iw_msixtbl)
		return -ENOMEM;

	rf->iw_qvlist = (struct irdma_qvlist_info *)
			(&rf->iw_msixtbl[rf->msix_count]);
	iw_qvlist = rf->iw_qvlist;
	iw_qvinfo = iw_qvlist->qv_info;
	iw_qvlist->num_vectors = rf->msix_count;
	if (rf->msix_count <= num_online_cpus())
		rf->msix_shared = true;
	else if (rf->msix_count > num_online_cpus() + 1)
		rf->msix_count = num_online_cpus() + 1;

	pmsix = rf->msix_entries;
	for (i = 0, ceq_idx = 0; i < rf->msix_count; i++, iw_qvinfo++) {
		rf->iw_msixtbl[i].idx = pmsix->entry;
		rf->iw_msixtbl[i].irq = pmsix->vector;
		rf->iw_msixtbl[i].cpu_affinity = ceq_idx;
		if (!i) { // 如果是第1个中断向量(i=0)且启用共享模式, 则将ceq_id自增1, 否则ceq_id为无效
			iw_qvinfo->aeq_idx = 0;
			if (rf->msix_shared)
				iw_qvinfo->ceq_idx = ceq_idx++;
			else
				iw_qvinfo->ceq_idx = IRDMA_Q_INVALID_IDX;
		} else {
			iw_qvinfo->aeq_idx = IRDMA_Q_INVALID_IDX;
			iw_qvinfo->ceq_idx = ceq_idx++;
		}
		iw_qvinfo->itr_idx = 3;
		iw_qvinfo->v_idx = rf->iw_msixtbl[i].idx;
		pmsix++;
	}

	return 0;
}

static int irdma_setup_ceq_0(struct irdma_pci_f *rf)
{
	struct irdma_ceq *iwceq;
	struct irdma_msix_vector *msix_vec;
	u32 i;
	int status = 0;
	u32 num_ceqs;

	num_ceqs = min(rf->msix_count, rf->sc_dev.hmc_fpm_misc.max_ceqs);
	rf->ceqlist = kcalloc(num_ceqs, sizeof(*rf->ceqlist), GFP_KERNEL);
	if (!rf->ceqlist) {
		status = -ENOMEM;
		goto exit;
	}

	iwceq = &rf->ceqlist[0];
	status = irdma_create_ceq(rf, iwceq, 0, &rf->default_vsi);
	if (status) {
		ibdev_dbg(&rf->iwdev->ibdev, "ERR: create ceq status = %d\n",
			  status);
		goto exit;
	}

	spin_lock_init(&iwceq->ce_lock);
	i = rf->msix_shared ? 0 : 1; // 如果是共享模式, 则ceq0使用第一个中断向量
	msix_vec = &rf->iw_msixtbl[i];
	iwceq->irq = msix_vec->irq;
	iwceq->msix_idx = msix_vec->idx;
	status = irdma_cfg_ceq_vector(rf, iwceq, 0, msix_vec); // 设置ceq_id为0
	if (status) {
		irdma_destroy_ceq(rf, iwceq);
		goto exit;
	}

	irdma_ena_intr(&rf->sc_dev, msix_vec->idx);
	rf->ceqs_count++;

exit:
	if (status && !rf->ceqs_count) {
		kfree(rf->ceqlist);
		rf->ceqlist = NULL;
		return status;
	}
	rf->sc_dev.ceq_valid = true;

	return 0;
}

irdma_create_ccq -> 创建用于控制操作的完成队列
    ccq->shadow_area.size = sizeof(struct irdma_cq_shadow_area)
    ccq->mem_cq.va = dma_alloc_coherent
    info.num_elem = IW_CCQ_SIZE -> 2048
    irdma_sc_ccq_init(dev->ccq, &info)
        cq->cq_type = IRDMA_CQ_TYPE_CQP
        IRDMA_RING_INIT(cq->cq_uk.cq_ring, info->num_elem)
        cq->cq_uk.polarity = true -> 极性位/可反转/有效位
    irdma_sc_ccq_create(dev->ccq, 0, true, true)
        irdma_sc_cq_create(ccq, scratch, check_overflow, post_sq) -> 创建完成队列
            ceq = cq->dev->ceq[cq->ceq_id]
             irdma_sc_add_cq_ctx(ceq, cq) -> 为 ceq 添加 cq ctx 跟踪
                ceq->reg_cq[ceq->reg_cq_size++] = cq
             ...
             set_64bit_val(wqe, 8, (uintptr_t)cq >> 1) -> 创建CQ时, 将CQ地址设置到CQP的WQE偏移8字节处

irdma_cfg_ceq_vector(rf, iwceq, 0, msix_vec) -> 为完成事件队列设置中断
    if (rf->msix_shared && !ceq_id)
        tasklet_setup(&rf->dpc_tasklet, irdma_dpc)
        request_irq(msix_vec->irq, irdma_irq_handler, 0,
            tasklet_schedule(&rf->dpc_tasklet)
    else
        tasklet_setup(&iwceq->dpc_tasklet, irdma_ceq_dpc)
            irdma_process_ceq(rf, iwceq)
                do cq = irdma_sc_process_ceq(dev, sc_ceq)
                    temp_cq = (struct irdma_sc_cq *)(unsigned long)(temp << 1) -> 处理CE时取回CQ
                    ceq->polarity ^= 1
                    cq_idx = irdma_sc_find_reg_cq(ceq, cq);
                    IRDMA_RING_MOVE_TAIL(ceq->ceq_ring)
                    irdma_sc_cq_ack(cq)
                        writel(cq->cq_uk.cq_id, cq->cq_uk.cq_ack_db)
                if (cq->cq_type == IRDMA_CQ_TYPE_IWARP)
                    irdma_iwarp_ce_handler(cq)
                        if (cq->ibcq.comp_handler)
                            cq->ibcq.comp_handler(&cq->ibcq, cq->ibcq.cq_context) -> ib_uverbs_comp_handler
                queue_work(rf->cqp_cmpl_wq, &rf->cqp_cmpl_work)

irdma_create_ceq(rf, iwceq, 0, &rf->default_vsi) -> 创建完成事件队列
    ceq_size = min(rf->sc_dev.hmc_info->hmc_obj[IRDMA_HMC_IW_CQ].cnt, dev->hw_attrs.max_hw_ceq_size)
    iwceq->mem.size = ALIGN(sizeof(struct irdma_ceqe) * ceq_size, IRDMA_CEQ_ALIGNMENT)
    iwceq->mem.va = dma_alloc_coherent(dev->hw->device, iwceq->mem.size, &iwceq->mem.pa, GFP_KERNEL)
    info.ceqe_base = iwceq->mem.va
    info.ceqe_pa = iwceq->mem.pa
    irdma_sc_ceq_init(&iwceq->sc_ceq, &info)
        ceq->ceqe_base = (struct irdma_ceqe *)info->ceqe_base
        ceq->polarity = 1
        IRDMA_RING_INIT(ceq->ceq_ring, ceq->elem_cnt)
    irdma_cqp_ceq_cmd(&rf->sc_dev, &iwceq->sc_ceq, IRDMA_OP_CEQ_CREATE) -> irdma_sc_ceq_create
    or irdma_sc_cceq_create(&iwceq->sc_ceq, 0)
        ret_code = irdma_sc_ceq_create(ceq, scratch, true)

#define IRDMA_GET_CURRENT_CEQ_ELEM(_ceq) \
	( \
		(_ceq)->ceqe_base[IRDMA_RING_CURRENT_TAIL((_ceq)->ceq_ring)].buf \
	)

case IRDMA_AE_RDMAP_ROE_BAD_LLP_CLOSE:
    if (qp->term_flags)
        break;
    if (atomic_inc_return(&iwqp->close_timer_started) == 1) {
        iwqp->hw_tcp_state = IRDMA_TCP_STATE_CLOSE_WAIT;
        if (iwqp->hw_tcp_state == IRDMA_TCP_STATE_CLOSE_WAIT &&
            iwqp->ibqp_state == IB_QPS_RTS) {
            irdma_next_iw_state(iwqp,
                        IRDMA_QP_STATE_CLOSING,
                        0, 0, 0);
            irdma_cm_disconn(iwqp);
        }
        irdma_schedule_cm_timer(iwqp->cm_node,
                    (struct irdma_puda_buf *)iwqp,
                    IRDMA_TIMER_TYPE_CLOSE,
                    1, 0);
    }
    break;

void *irdma_sc_process_ceq(struct irdma_sc_dev *dev, struct irdma_sc_ceq *ceq)
{
	u64 temp;
	__le64 *ceqe;
	struct irdma_sc_cq *cq = NULL;
	struct irdma_sc_cq *temp_cq;
	u8 polarity;
	u32 cq_idx;
	unsigned long flags;

	do { // 处理CEQ中所有的CEQE(循环/遍历, CEQ的环形BUF)
		cq_idx = 0;
		ceqe = IRDMA_GET_CURRENT_CEQ_ELEM(ceq);
		get_64bit_val(ceqe, 0, &temp);
		polarity = (u8)FIELD_GET(IRDMA_CEQE_VALID, temp);
		if (polarity != ceq->polarity) // 当前CEQE极性与CEQ极性不同则认为该CEQE无效, 直接返回NULL
			return NULL;

		temp_cq = (struct irdma_sc_cq *)(unsigned long)(temp << 1);
		if (!temp_cq) {
			cq_idx = IRDMA_INVALID_CQ_IDX;
			IRDMA_RING_MOVE_TAIL(ceq->ceq_ring);

			if (!IRDMA_RING_CURRENT_TAIL(ceq->ceq_ring))
				ceq->polarity ^= 1;
			continue;
		}

		cq = temp_cq;
		if (ceq->reg_cq) {
			spin_lock_irqsave(&ceq->req_cq_lock, flags);
			cq_idx = irdma_sc_find_reg_cq(ceq, cq);
			spin_unlock_irqrestore(&ceq->req_cq_lock, flags);
		}

		IRDMA_RING_MOVE_TAIL(ceq->ceq_ring);
		if (!IRDMA_RING_CURRENT_TAIL(ceq->ceq_ring))
			ceq->polarity ^= 1; // TAIL指针重新回到0时(绕圈), 将CEQ上的极性翻转
	} while (cq_idx == IRDMA_INVALID_CQ_IDX);

	if (cq)
		irdma_sc_cq_ack(cq);
	return cq;
}

void irdma_sc_cleanup_ceqes(struct irdma_sc_cq *cq, struct irdma_sc_ceq *ceq)
{
	struct irdma_sc_cq *next_cq;
	u8 ceq_polarity = ceq->polarity; // 获取当前CEQ极性
	__le64 *ceqe;
	u8 polarity;
	u64 temp;
	int next;
	u32 i;

	next = IRDMA_RING_GET_NEXT_TAIL(ceq->ceq_ring, 0);

	for (i = 1; i <= IRDMA_RING_SIZE(*ceq); i++) {
		ceqe = IRDMA_GET_CEQ_ELEM_AT_POS(ceq, next);

		get_64bit_val(ceqe, 0, &temp); // 获取CEQE的全部内容(64bit/8字节)
		polarity = (u8)FIELD_GET(IRDMA_CEQE_VALID, temp); // 获取CEQE的极性
		if (polarity != ceq_polarity)
			return; // 无效CEQE直接退出

		next_cq = (struct irdma_sc_cq *)(unsigned long)(temp << 1);
		if (cq == next_cq)
			set_64bit_val(ceqe, 0, temp & IRDMA_CEQE_VALID); // IRDMA_CEQE_VALID(第63bit, 默认为0), &(与):忽略最高的第63bit, 将CQ_Context_Value清零, 使CEQE无效, 在处理CEQ的函数中, 获取不到CQ(设置cq_idx = IRDMA_INVALID_CQ_IDX), 即停止处理CEQ
        
		next = IRDMA_RING_GET_NEXT_TAIL(ceq->ceq_ring, i);
		if (!next)
			ceq_polarity ^= 1;
	}
}

/**
 * irdma_destroy_cq - destroy cq
 * @ib_cq: cq pointer
 * @udata: user data
 */
static int irdma_destroy_cq(struct ib_cq *ib_cq, struct ib_udata *udata)
{
	struct irdma_device *iwdev = to_iwdev(ib_cq->device);
	struct irdma_cq *iwcq = to_iwcq(ib_cq);
	struct irdma_sc_cq *cq = &iwcq->sc_cq;
	struct irdma_sc_dev *dev = cq->dev;
	struct irdma_sc_ceq *ceq = dev->ceq[cq->ceq_id];
	struct irdma_ceq *iwceq = container_of(ceq, struct irdma_ceq, sc_ceq);
	unsigned long flags;

	spin_lock_irqsave(&iwcq->lock, flags);
	if (!list_empty(&iwcq->cmpl_generated))
		irdma_remove_cmpls_list(iwcq);
	if (!list_empty(&iwcq->resize_list))
		irdma_process_resize_list(iwcq, iwdev, NULL);
	spin_unlock_irqrestore(&iwcq->lock, flags);

	irdma_cq_rem_ref(ib_cq);
	wait_for_completion(&iwcq->free_cq); // 等待

	irdma_cq_wq_destroy(iwdev->rf, cq);

	spin_lock_irqsave(&iwceq->ce_lock, flags);
	irdma_sc_cleanup_ceqes(cq, ceq);
	spin_unlock_irqrestore(&iwceq->ce_lock, flags);
	irdma_cq_free_rsrc(iwdev->rf, iwcq);

	return 0;
}

IB设备
struct ib_device {
    int			      num_comp_vectors; -> IB：添加 CQ comp_vector 支持，向 struct ib_device 添加 num_comp_vectors 成员，并扩展 ib_create_cq() 以传入 comp_vector 参数——这与用户空间 libibverbs API 并行。 更新所有硬件驱动程序，将 num_comp_vectors 设置为 1，并使所有 ULP 将 comp_vector 值传递为 0。 将 num_comp_vectors 的值传递到用户空间，而不是硬编码值 1。我们需要多个 CQ 事件向量支持（通过 MSI-X 或类似的适配器，可以生成多个中断），但不清楚我们需要多少个向量， 或者我们想要如何处理策略问题，例如如何决定使用哪个向量或如何设置中断关联。 该补丁对于实验很有用，因为在更新驱动程序以支持多个向量时不需要进行核心更改，而且我们知道无论如何我们至少希望进行这些更改
}

fi_eq_open(fabric, &eq_attr, &eq, NULL)
struct fi_eq_attr {
	size_t               size;      /* # entries for EQ */
	uint64_t             flags;     /* operation flags */
	enum fi_wait_obj     wait_obj;  /* requested wait object */
	int                  signaling_vector; /* interrupt affinity 中断亲和性(通知信号向量) */
	struct fid_wait     *wait_set;  /* optional wait set, deprecated */
};    
*signaling_vector* ：如果设置了 FI_AFFINITY 标志，则表示与 EQ 关联的中断应针对的逻辑 CPU 编号（0..最大 CPU - 1）。此字段应被视为对提供程序的提示，如果提供程序不支持中断亲和性，则可以忽略此字段
    
fi_cq_open
int vrb_cq_open(struct fid_domain *domain_fid, struct fi_cq_attr *attr,
		   struct fid_cq **cq_fid, void *context)
{
	struct vrb_cq *cq;
	struct vrb_domain *domain =
		container_of(domain_fid, struct vrb_domain,
			     util_domain.domain_fid);
	size_t size;
	int ret;
	struct fi_cq_attr tmp_attr = *attr;
	int comp_vector = 0;

	cq = calloc(1, sizeof(*cq));
	if (!cq)
		return -FI_ENOMEM;

	/* verbs uses its own implementation of wait objects for CQ */
	tmp_attr.wait_obj = FI_WAIT_NONE;
	ret = ofi_cq_init(&vrb_prov, domain_fid, &tmp_attr, &cq->util_cq,
			  vrb_cq_progress, context);
	if (ret)
		goto err1;

	switch (attr->wait_obj) {
	case FI_WAIT_UNSPEC:
		cq->wait_obj = FI_WAIT_FD;
		break;
	case FI_WAIT_FD:
	case FI_WAIT_POLLFD:
	case FI_WAIT_NONE:
		cq->wait_obj = attr->wait_obj;
		break;
	default:
		ret = -FI_ENOSYS;
		goto err2;
	}

	if (attr->flags & FI_AFFINITY) { 
		if (attr->signaling_vector < 0 ||
		    attr->signaling_vector > domain->verbs->num_comp_vectors)  { // 信号向量最大值边界

			VRB_WARN(FI_LOG_CQ,
				   "Invalid value for the CQ attribute signaling_vector: %d\n",
				   attr->signaling_vector);
			ret = -FI_EINVAL;
			goto err2;
		}
		comp_vector = attr->signaling_vector; // 如果启用完成事件中断与CPU亲和性, 则设置完成向量为传入的信号向量值
	}

	if (cq->wait_obj != FI_WAIT_NONE) {
		cq->channel = ibv_create_comp_channel(domain->verbs);
		if (!cq->channel) {
			ret = -errno;
			VRB_WARN(FI_LOG_CQ,
				   "Unable to create completion channel\n");
			goto err2;
		}

		ret = fi_fd_nonblock(cq->channel->fd);
		if (ret)
			goto err3;

		if (fd_signal_init(&cq->signal)) {
			ret = -errno;
			goto err3;
		}
	}

	size = attr->size ? attr->size : VERBS_DEF_CQ_SIZE;

	/*
	 * Verbs may throw an error if CQ size exceeds ibv_device_attr->max_cqe.
	 * OFI doesn't expose CQ size to the apps because it's better to fix the
	 * issue in the provider than the app dealing with it. The fix is to
	 * open multiple verbs CQs and load balance "MSG EP to CQ binding"* among
	 * them to avoid any CQ overflow.
	 * Something like:
	 * num_qp_per_cq = ibv_device_attr->max_cqe / (qp_send_wr + qp_recv_wr)
	 */
	cq->cq = ibv_create_cq(domain->verbs, size, cq, cq->channel,
			       comp_vector); // 创建CQ时将完成向量(亲和性)作为参数
	if (!cq->cq) {
		ret = -errno;
		VRB_WARN(FI_LOG_CQ, "Unable to create verbs CQ\n");
		goto err3;
	}

	if (cq->channel) {
		ret = ibv_req_notify_cq(cq->cq, 0);
		if (ret) {
			VRB_WARN(FI_LOG_CQ,
				   "ibv_req_notify_cq failed\n");
			goto err4;
		}
	}

	cq->flags |= attr->flags;
	cq->wait_cond = attr->wait_cond;
	/* verbs uses its own ops for CQ */
	cq->util_cq.cq_fid.fid.ops = &vrb_cq_fi_ops;
	cq->util_cq.cq_fid.ops = &vrb_cq_ops;

	// slist_init(&cq->saved_wc_list);
	dlist_init(&cq->xrc.srq_list);

	ofi_atomic_initialize32(&cq->nevents, 0);

	*cq_fid = &cq->util_cq.cq_fid;
	return 0;

err4:
	ibv_destroy_cq(cq->cq);
err3:
	if (cq->channel)
		ibv_destroy_comp_channel(cq->channel);
err2:
	ofi_cq_cleanup(&cq->util_cq);
err1:
	free(cq);
	return ret;
}