rdma cq的同步事件通知机制_rdvmqg

struct ibv_comp_channel *ibv_create_comp_channel(struct ibv_context *context);
int ibv_destroy_comp_channel(struct ibv_comp_channel *channel);
int ibv_get_cq_event(struct ibv_comp_channel *channel,
		     struct ibv_cq **cq, void **cq_context);
void ibv_ack_cq_events(struct ibv_cq *cq, unsigned int nevents);
int ibv_req_notify_cq(struct ibv_cq *cq, int solicited_only);
1
2
3
4
5
6

使用例程：https://www.rdmamojo.com/2013/03/16/ibv_ack_cq_events/

内核rxe驱动，rxe_req_notify_cq函数内容
原理分析：

static int rxe_req_notify_cq(struct ib_cq *ibcq, enum ib_cq_notify_flags flags)
{
	struct rxe_cq *cq = to_rcq(ibcq);
	unsigned long irq_flags;
	int ret = 0;
	int empty;

	spin_lock_irqsave(&cq->cq_lock, irq_flags);
	if (cq->notify != IB_CQ_NEXT_COMP)//2
		cq->notify = flags & IB_CQ_SOLICITED_MASK;//3

	empty = queue_empty(cq->queue, QUEUE_TYPE_FROM_DRIVER);

	if ((flags & IB_CQ_REPORT_MISSED_EVENTS) && !empty)
		ret = 1;

	spin_unlock_irqrestore(&cq->cq_lock, irq_flags);

	return ret;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

这里最重要的是去设置cq->notify的值，分两种情况，
1)如果flags给0，那么就意味则只要有cqe就可以通知上层应用程序
2)有solicited_only类型的cqe才通知上层

设置好cq->notify的值以后，就看cqe什么时候生成了。

在rxe_cq.c里，只要有cqe生成就会调用该函数

int rxe_cq_post(struct rxe_cq *cq, struct rxe_cqe *cqe, int solicited)
{
	struct ib_event ev;
	unsigned long flags;
	int full;
	void *addr;

	spin_lock_irqsave(&cq->cq_lock, flags);

	full = queue_full(cq->queue, QUEUE_TYPE_TO_CLIENT);
	if (unlikely(full)) {
		spin_unlock_irqrestore(&cq->cq_lock, flags);
		if (cq->ibcq.event_handler) {
			ev.device = cq->ibcq.device;
			ev.element.cq = &cq->ibcq;
			ev.event = IB_EVENT_CQ_ERR;
			cq->ibcq.event_handler(&ev, cq->ibcq.cq_context);
		}

		return -EBUSY;
	}

	addr = queue_producer_addr(cq->queue, QUEUE_TYPE_TO_CLIENT);
	memcpy(addr, cqe, sizeof(*cqe));

	queue_advance_producer(cq->queue, QUEUE_TYPE_TO_CLIENT);

	spin_unlock_irqrestore(&cq->cq_lock, flags);

	if ((cq->notify == IB_CQ_NEXT_COMP) ||
	    (cq->notify == IB_CQ_SOLICITED && solicited)) {
		cq->notify = 0;
		tasklet_schedule(&cq->comp_task);
	}

	return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

其中最重要的是去调度tasklet_schedule(&cq->comp_task);然后设置cq->notify为0，所以为什么说，
ibv_req_notify_cq函数要重复的调用。

调度会调用到rxe_send_complete函数，通过cq->ibcq.comp_handler(&cq->ibcq, cq->ibcq.cq_context);
通知应用程序，有cqe产生了。

static void rxe_send_complete(struct tasklet_struct *t)
{
	struct rxe_cq *cq = from_tasklet(cq, t, comp_task);
	unsigned long flags;

	spin_lock_irqsave(&cq->cq_lock, flags);
	if (cq->is_dying) {
		spin_unlock_irqrestore(&cq->cq_lock, flags);
		return;
	}
	spin_unlock_irqrestore(&cq->cq_lock, flags);

	cq->ibcq.comp_handler(&cq->ibcq, cq->ibcq.cq_context);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

这个时候，ibv_get_cq_event函数的read就会返回，

LATEST_SYMVER_FUNC(ibv_get_cq_event, 1_1, "IBVERBS_1.1",
		   int,
		   struct ibv_comp_channel *channel,
		   struct ibv_cq **cq, void **cq_context)
{
	struct ib_uverbs_comp_event_desc ev;

	if (read(channel->fd, &ev, sizeof ev) != sizeof ev)
		return -1;

	*cq         = (struct ibv_cq *) (uintptr_t) ev.cq_handle;
	*cq_context = (*cq)->cq_context;

	get_ops((*cq)->context)->cq_event(*cq);

	return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

随后调用ibv_ack_cq_events确认已经收到了该事件。

LATEST_SYMVER_FUNC(ibv_ack_cq_events, 1_1, "IBVERBS_1.1",
		   void,
		   struct ibv_cq *cq, unsigned int nevents)
{
	pthread_mutex_lock(&cq->mutex);
	cq->comp_events_completed += nevents;
	pthread_cond_signal(&cq->cond);
	pthread_mutex_unlock(&cq->mutex);
}
1
2
3
4
5
6
7
8
9

这里其实就是对comp_events_completed 值进行累加，在销毁cq时，有个判断

我想，如果不调用ibv_ack_cq_events的话，在销毁cq时，应该是会阻塞的，可以自己写代码验证下。