rpma: commit at least 1000 cq events at a time

src/cq.c

@@ -19,10 +19,13 @@
 #include "cmocka_alloc.h"
 #endif
 
+#define RPMA_MAX_UNACK_CQE 1000
+
 struct rpma_cq {
 	struct ibv_comp_channel *channel; /* completion channel */
 	bool shared_comp_channel; /* completion channel is shared */
 	struct ibv_cq *cq; /* completion queue */
+	unsigned unack_cqe; /* number of unack completion events */
 };
 
 /* internal librpma API */
@@ -94,6 +97,8 @@ rpma_cq_new(struct ibv_context *ibv_ctx, int cqe,
 	(*cq_ptr)->channel = channel;
 	(*cq_ptr)->shared_comp_channel = (shared_channel != NULL);
 	(*cq_ptr)->cq = cq;
+	(*cq_ptr)->unack_cqe = 0;
+
 
 	return 0;
 
@@ -124,6 +129,10 @@ rpma_cq_delete(struct rpma_cq **cq_ptr)
 	if (cq == NULL)
 		return ret;
 
+	if (cq->unack_cqe)
+		(void) ibv_ack_cq_events(cq->cq, cq->unack_cqe);
+	cq->unack_cqe = 0;
+
 	errno = ibv_destroy_cq(cq->cq);
 	if (errno) {
 		RPMA_LOG_ERROR_WITH_ERRNO(errno, "ibv_destroy_cq()");
@@ -175,19 +184,29 @@ rpma_cq_wait(struct rpma_cq *cq)
 	if (cq->shared_comp_channel)
 		return RPMA_E_SHARED_CHANNEL;
 
+	/*
+	 * ACK the collected CQ event.
+	 */
+	if (cq->unack_cqe >= RPMA_MAX_UNACK_CQE) {
+		ibv_ack_cq_events(cq->cq, cq->unack_cqe);
+		cq->unack_cqe = 0;
+	}
+
 	/* wait for the completion event */
 	struct ibv_cq *ev_cq;	/* unused */
 	void *ev_ctx;		/* unused */
 	if (ibv_get_cq_event(cq->channel, &ev_cq, &ev_ctx))
 		return RPMA_E_NO_COMPLETION;
 
+	++cq->unack_cqe;
+
 	/*
 	 * ACK the collected CQ event.
 	 *
 	 * XXX for performance reasons, it may be beneficial to ACK more than
 	 * one CQ event at the same time.
 	 */
-	ibv_ack_cq_events(cq->cq, 1 /* # of CQ events */);
+	// ibv_ack_cq_events(cq->cq, 1 /* # of CQ events */);
 
 	/* request for the next event on the CQ channel */
 	errno = ibv_req_notify_cq(cq->cq, 0 /* all completions */);

tests/unit/common/mocks-ibverbs.c

@@ -25,6 +25,7 @@ struct ibv_cq Ibv_cq_unknown;
 struct ibv_qp Ibv_qp;
 struct ibv_mr Ibv_mr;
 
+static unsigned unack_cqe = 0;
 /*
  * ibv_query_device -- ibv_query_device() mock
  */
@@ -205,6 +206,7 @@ ibv_get_cq_event(struct ibv_comp_channel *channel, struct ibv_cq **cq,
 	if (!errno) {
 		*cq = mock_type(struct ibv_cq *);
 		*cq_context = NULL;
+		unack_cqe++;
 		return 0;
 	}
 
@@ -217,8 +219,10 @@ ibv_get_cq_event(struct ibv_comp_channel *channel, struct ibv_cq **cq,
 void
 ibv_ack_cq_events(struct ibv_cq *cq, unsigned nevents)
 {
-	check_expected_ptr(cq);
-	assert_int_equal(nevents, 1);
+	// check_expected_ptr(cq);
+	if (unack_cqe)
+		assert_int_equal(nevents, unack_cqe);
+	unack_cqe = 0;
 }
 
 /*

tests/unit/cq/cq-common.c

@@ -64,6 +64,7 @@ teardown__cq_delete(void **cq_ptr)
 	struct rpma_cq *cq = cstate->cq;
 
 	/* configure mocks */
+	will_return_maybe(ibv_ack_cq_events, MOCK_OK);
 	will_return(ibv_destroy_cq, MOCK_OK);
 	if (!cstate->shared_channel)
 		will_return(ibv_destroy_comp_channel, MOCK_OK);

tests/unit/cq/cq-wait.c

@@ -76,7 +76,7 @@ wait__req_notify_cq_ERRNO(void **cq_ptr)
 	expect_value(ibv_get_cq_event, channel, MOCK_COMP_CHANNEL);
 	will_return(ibv_get_cq_event, MOCK_OK);
 	will_return(ibv_get_cq_event, MOCK_IBV_CQ);
-	expect_value(ibv_ack_cq_events, cq, MOCK_IBV_CQ);
+	// expect_value(ibv_ack_cq_events, cq, MOCK_IBV_CQ);
 	expect_value(ibv_req_notify_cq_mock, cq, MOCK_IBV_CQ);
 	will_return(ibv_req_notify_cq_mock, MOCK_ERRNO);
 
@@ -100,7 +100,7 @@ wait__success(void **cq_ptr)
 	expect_value(ibv_get_cq_event, channel, MOCK_COMP_CHANNEL);
 	will_return(ibv_get_cq_event, MOCK_OK);
 	will_return(ibv_get_cq_event, MOCK_IBV_CQ);
-	expect_value(ibv_ack_cq_events, cq, MOCK_IBV_CQ);
+	// expect_value(ibv_ack_cq_events, cq, MOCK_IBV_CQ);
 	expect_value(ibv_req_notify_cq_mock, cq, MOCK_IBV_CQ);
 	will_return(ibv_req_notify_cq_mock, MOCK_OK);