FEAT: rpma_conn_wait() performance optimization

[grom72]

FEAT: rpma_conn_wait() performance optimization

Rationale

Having rpma_conn_wait() a part of the librpma API we can implement a similar solution as the one described in #1697 but also give the librpma user the possibility to configure MAX_UNACK_CQE via struct rpma_conn_cfg.

Description

The struct rpma_conn shall be extended with two fields (counters) unsigned int unack_cqe and unsigned int unack_rcqe and set to 0 in rpma_conn_new().

These counters shall be incremented inside rpma_conn_wait() in a similar way as it is described in #1697:

...
	if (conn->cq && (rpma_cq_get_ibv_cq(conn->cq) == ev_cq)) {
		*cq = conn->cq;
		++conn->unack_cqe;
		if (is_rcq)
			*is_rcq = false;
	} else if (conn->rcq && (rpma_cq_get_ibv_cq(conn->rcq) == ev_cq)) {
		*cq = conn->rcq;
		++conn->unack_rcqe;
		if (is_rcq)
			*is_rcq = true;
	} else {
...

ibv_ack_cq_events() shall be called inside rpma_conn_delete() before the rpma_cq_delet() call is made.

...
	if (conn->unack_rcqe)
		(void) ibv_ack_cq_events(conn->rcq, conn->unack_rcqe);
	ret = rpma_cq_delete(&conn->rcq);
	if (ret)
		goto err_rpma_cq_delete;

	if (conn->unack_cqe)
		(void) ibv_ack_cq_events(conn->rcq, conn->unack_cqe);
	ret = rpma_cq_delete(&conn->cq);
	if (ret)
		goto err_destroy_id;
...

To have the possibility dynamically configure the number of events in one ibv_ack_cq_events() call the rpma_conn_cfg shall be extended with int max_unack_cqe

struct rpma_conn_cfg {
...
	unsigned int max_unack_cqe; /* max unacked completion events */
};

This value must be propagated to a new field unsigned int max_unack_cqe; in struct rpma_conn_req and in struct rpma_conn as part of connection setup and establishing part: rpma_conn_req_new(), rpma_conn_req_connect(), rpma_conn_req_accept() and rpma_conn_req_connect_active()

Having all these in place we can adjust rpma_conn_wait() to follow the new parameter instead of having fixed RPMA_MAX_INACK_CQE value (please observe that ibv_ack_cq_events() operation is moved before ibv_get_cq_event()):

int
rpma_conn_wait(struct rpma_conn *conn, struct rpma_cq **cq, bool *is_rcq)
{
...
	/*
	 * ACK the collected CQ event.
	 */
	if (conn->unack_rcqe >= conn->max_unack_cqe) {
		ibv_ack_cq_events(conn->rcq, conn->unack_rcqe);
		conn->unack_rcqe= 0;
	}
	if (conn->unack_cqe >= conn->max_unack_cqe) {
		ibv_ack_cq_events(conn->cq, conn->unack_cqe);
		conn->unack_cqe= 0;
	}
	if (ibv_get_cq_event(conn->channel, &ev_cq, &ev_ctx))
		return RPMA_E_NO_COMPLETION;
...