prov/efa: Add support for blocking CQ read in RDM

This adds support for `fi_cq_sread{,from}`, `fi_control` and `fi_signal`
for the `efa` fabric. This also expands EFA's `fi_trywait()` to handle
wait objects for Util CQs.

Since the endpoint needs to progress in order to generate completions to
read from, an exponential backoff strategy is employed to poll by
driving progress periodically without excessively spinning the CPU for
higher latency scenarios. The `timeout` specified by the user is used to
cap the total time spent waiting for completions.

A CQ requested with FI_WAIT_FD indicates the app intends to drive
progress itself as needed, since EFA advertises FI_PROGRESS_MANUAL. In
this scenario, sreadfrom will block without periodically driving
progress.

Signed-off-by: Darryl Abbate <[email protected]>

Author: darrylabbate

prov/efa/src/efa_fabric.c

@@ -20,6 +20,7 @@
 
 #include "efa_cq.h"
 #include "efa_prov_info.h"
+#include "rdm/efa_rdm_cq.h"
 #ifdef EFA_PERF_ENABLED
 const char *efa_perf_counters_str[] = {
 	EFA_PERF_FOREACH(OFI_STR)
@@ -63,16 +64,28 @@ static int efa_fabric_close(fid_t fid)
 static int efa_trywait(struct fid_fabric *fabric, struct fid **fids, int count)
 {
 	struct efa_cq *efa_cq;
+	struct util_cq *util_cq;
+	struct efa_rdm_cq *rdm_cq;
 	struct util_wait *wait;
 	int ret, i;
 
 	for (i = 0; i < count; i++) {
 		if (fids[i]->fclass == FI_CLASS_CQ) {
-			/* Use EFA-specific CQ trywait */
-			efa_cq = container_of(fids[i], struct efa_cq, util_cq.cq_fid.fid);
-			ret = efa_cq_trywait(efa_cq);
-			if (ret)
-				return ret;
+			util_cq = container_of(fids[i], struct util_cq, cq_fid.fid);
+			
+			/* RDM CQs use util wait objects, not hardware CQ events */
+			if (util_cq->wait) {
+				rdm_cq = container_of(util_cq, struct efa_rdm_cq, efa_cq.util_cq);
+				ret = efa_rdm_cq_trywait(rdm_cq);
+				if (ret)
+					return ret;
+			} else {
+				/* Use EFA-specific CQ trywait for non-RDM CQs */
+				efa_cq = container_of(fids[i], struct efa_cq, util_cq.cq_fid.fid);
+				ret = efa_cq_trywait(efa_cq);
+				if (ret)
+					return ret;
+			}
 		} else {
 			/* Use generic util trywait logic for non-CQ types */
 			switch (fids[i]->fclass) {

prov/efa/src/rdm/efa_rdm_cq.c

@@ -8,6 +8,7 @@
 #include "efa_av.h"
 #include "efa_cntr.h"
 #include "efa_rdm_pke_cmd.h"
+#include "efa_rdm_peer.h"
 #include "efa_rdm_pke_utils.h"
 #include "efa_rdm_pke_nonreq.h"
 #include "efa_rdm_tracepoint.h"
@@ -22,6 +23,28 @@ const char *efa_rdm_cq_strerror(struct fid_cq *cq_fid, int prov_errno,
 		: efa_strerror(prov_errno);
 }
 
+static
+int efa_rdm_cq_control(struct fid *fid, int command, void *arg)
+{
+	const struct util_cq *util_cq = container_of(fid, struct util_cq, cq_fid.fid);
+
+	switch (command) {
+	case FI_GETWAIT:
+		if (!util_cq->wait || util_cq->wait->wait_obj != FI_WAIT_FD)
+			return -FI_ENODATA;
+		*(int *) arg = container_of(util_cq->wait, struct util_wait_fd, util_wait)->epoll_fd;
+		break;
+	case FI_GETWAITOBJ:
+		*(enum fi_wait_obj *) arg = util_cq->wait
+			? util_cq->wait->wait_obj
+			: FI_WAIT_NONE;
+		break;
+	default:
+		return -FI_ENOSYS;
+	}
+	return FI_SUCCESS;
+}
+
 /**
  * @brief close a CQ of EFA RDM endpoint
  *
@@ -69,7 +92,7 @@ static struct fi_ops efa_rdm_cq_fi_ops = {
 	.size = sizeof(struct fi_ops),
 	.close = efa_rdm_cq_close,
 	.bind = fi_no_bind,
-	.control = fi_no_control,
+	.control = efa_rdm_cq_control,
 	.ops_open = fi_no_ops_open,
 };
 
@@ -116,6 +139,9 @@ void efa_rdm_cq_proc_ibv_recv_rdma_with_imm_completion(
 		ret = ofi_cq_write(target_cq, NULL, flags, len, NULL, imm_data, 0);
 	}
 
+	if (OFI_LIKELY(!ret && target_cq->wait))
+		target_cq->wait->signal(target_cq->wait);
+
 	if (OFI_UNLIKELY(ret)) {
 		EFA_WARN(FI_LOG_CQ,
 			"Unable to write a cq entry for remote for RECV_RDMA operation: %s\n",
@@ -854,14 +880,113 @@ static ssize_t efa_rdm_cq_readfrom(struct fid_cq *cq_fid, void *buf, size_t coun
 	return ret;
 }
 
+static void efa_rdm_cq_progress(struct util_cq *cq);
+
+/**
+ * @brief Blocking CQ read with source address
+ *
+ * @param[in]     cq_fid     CQ file descriptor
+ * @param[out]    buf        Buffer for completion entries
+ * @param[in]     count      Maximum entries to read
+ * @param[out]    src_addr   Source addresses (optional)
+ * @param[in]     cond       Wait condition (threshold)
+ * @param[in]     timeout    Timeout in milliseconds
+ * @return        Number of entries read or negative error code
+ */
+static ssize_t efa_rdm_cq_sreadfrom(struct fid_cq *cq_fid, void *buf, size_t count,
+				    fi_addr_t *src_addr, const void *cond, int timeout)
+{
+	struct efa_cq *efa_cq = container_of(cq_fid, struct efa_cq, util_cq.cq_fid);
+	struct util_cq *util_cq = &efa_cq->util_cq;
+	const uint64_t endtime = ofi_timeout_time(timeout);
+	ssize_t ret;
+	size_t num_completions = 0, threshold = 1;
+	int wait_time = 1;
+	uint8_t *buffer = buf;
+
+	/* fi_cq_sread[from] calls are invalid w/ FI_WAIT_NONE */
+	if (OFI_UNLIKELY(efa_cq->wait_obj == FI_WAIT_NONE))
+		return -FI_EINVAL;
+
+	if (OFI_UNLIKELY(!count))
+		return -FI_EINVAL;
+
+	/* Require wait object for blocking operations */
+	if (OFI_UNLIKELY(!util_cq->wait))
+		return -FI_ENOSYS;
+
+	/* Handle threshold condition */
+	if (OFI_UNLIKELY(efa_cq->wait_cond == FI_CQ_COND_THRESHOLD && cond))
+		threshold = MAX(MIN(*(const size_t *) cond, count), 1);
+
+	while (num_completions < threshold) {
+		/* Try non-blocking read first */
+		ret = efa_rdm_cq_readfrom(cq_fid, buffer, count - num_completions,
+					  src_addr ? src_addr + num_completions : NULL);
+		
+		if (OFI_LIKELY(ret > 0)) {
+			num_completions += ret;
+			buffer += (ptrdiff_t) ret * efa_cq->entry_size;
+			wait_time = 1;
+			continue;
+		}
+
+		if (OFI_UNLIKELY(ret != 0 && ret != -FI_EAGAIN))
+			return num_completions ? (ssize_t) num_completions : ret;
+
+		/* Handle timeout */
+		if (OFI_UNLIKELY(ofi_adjust_timeout(endtime, &timeout)))
+			return num_completions ? (ssize_t) num_completions : -FI_EAGAIN;
+
+		/* Check for wakeup signal */
+		if (OFI_UNLIKELY(ofi_atomic_get32(&util_cq->wakeup))) {
+			ofi_atomic_set32(&util_cq->wakeup, 0);
+			return num_completions ? (ssize_t) num_completions : -FI_EAGAIN;
+		}
+
+		/* Progress then wait */
+		efa_rdm_cq_progress(util_cq);
+		
+		/* Check if progress produced completions before blocking */
+		if (OFI_LIKELY(!ofi_cirque_isempty(util_cq->cirq)))
+			continue;
+		
+		if (efa_cq->wait_obj == FI_WAIT_FD) {
+			ofi_wait(&util_cq->wait->wait_fid, timeout);
+		} else {
+			/* Wait with exponential backoff, capped at remaining timeout */
+			ofi_wait(&util_cq->wait->wait_fid, timeout >= 0 ? MIN(wait_time, timeout) : wait_time);
+			wait_time *= 2;
+		}
+	}
+
+	return (ssize_t) num_completions;
+}
+
+/**
+ * @brief Blocking CQ read for EFA RDM
+ *
+ * @param[in]     cq_fid     CQ file descriptor
+ * @param[out]    buf        Buffer for completion entries
+ * @param[in]     count      Maximum entries to read
+ * @param[in]     cond       Wait condition (unused)
+ * @param[in]     timeout    Timeout in milliseconds
+ * @return        Number of entries read or negative error code
+ */
+static ssize_t efa_rdm_cq_sread(struct fid_cq *cq_fid, void *buf, size_t count,
+				const void *cond, int timeout)
+{
+	return efa_rdm_cq_sreadfrom(cq_fid, buf, count, NULL, cond, timeout);
+}
+
 static struct fi_ops_cq efa_rdm_cq_ops = {
 	.size = sizeof(struct fi_ops_cq),
 	.read = ofi_cq_read,
 	.readfrom = efa_rdm_cq_readfrom,
 	.readerr = ofi_cq_readerr,
-	.sread = fi_no_cq_sread,
-	.sreadfrom = fi_no_cq_sreadfrom,
-	.signal = fi_no_cq_signal,
+	.sread = efa_rdm_cq_sread,
+	.sreadfrom = efa_rdm_cq_sreadfrom,
+	.signal = ofi_cq_signal,
 	.strerror = efa_rdm_cq_strerror,
 };
 
@@ -903,6 +1028,61 @@ static void efa_rdm_cq_progress(struct util_cq *cq)
 	ofi_genlock_unlock(&cq->ep_list_lock);
 }
 
+/**
+ * @brief Try to wait on RDM CQ - check if ready for blocking wait
+ *
+ * Called only when util_cq->wait exists
+ *
+ * @param[in]	cq	RDM CQ to check
+ * @return	FI_SUCCESS if ready to wait, -FI_EAGAIN if completions available
+ */
+int efa_rdm_cq_trywait(struct efa_rdm_cq *cq)
+{
+	struct util_cq *util_cq = &cq->efa_cq.util_cq;
+
+	/* Drive progress */
+	fi_cq_read(&util_cq->cq_fid, NULL, 0);
+	
+	/* Check if completions available */
+	if (!ofi_cirque_isempty(util_cq->cirq))
+		return -FI_EAGAIN;
+	
+	/* Use wait object's trywait (guaranteed to exist by caller) */
+	return util_cq->wait->wait_try(util_cq->wait);
+}
+
+/**
+ * @brief Verify CQ wait object attributes
+ *
+ * @param[in]	attr	CQ attributes to verify
+ * @return	FI_SUCCESS if valid, -FI_EINVAL if unsupported
+ */
+static int efa_rdm_cq_verify_wait_attr(const struct fi_cq_attr *attr)
+{
+	switch (attr->wait_obj) {
+	case FI_WAIT_NONE:
+	case FI_WAIT_UNSPEC:
+	case FI_WAIT_FD:
+		break;
+	default:
+		EFA_WARN(FI_LOG_CQ, "Unsupported wait object: %s\n",
+			 fi_tostr(&attr->wait_obj, FI_TYPE_WAIT_OBJ));
+		return -FI_EINVAL;
+	}
+	
+	switch (attr->wait_cond) {
+	case FI_CQ_COND_NONE:
+	case FI_CQ_COND_THRESHOLD:
+		break;
+	default:
+		EFA_WARN(FI_LOG_CQ, "Unsupported wait condition: %s\n",
+			 fi_tostr(&attr->wait_cond, FI_TYPE_CQ_WAIT_COND));
+		return -FI_EINVAL;
+	}
+	
+	return FI_SUCCESS;
+}
+
 /**
  * @brief create a CQ for EFA RDM provider
  *
@@ -926,8 +1106,9 @@ int efa_rdm_cq_open(struct fid_domain *domain, struct fi_cq_attr *attr,
 	struct fi_peer_cq_context peer_cq_context = {0};
 	struct fi_efa_cq_init_attr efa_cq_init_attr = {0};
 
-	if (attr->wait_obj != FI_WAIT_NONE)
-		return -FI_ENOSYS;
+	ret = efa_rdm_cq_verify_wait_attr(attr);
+	if (ret)
+		return ret;
 
 	cq = calloc(1, sizeof(*cq));
 	if (!cq)
@@ -940,12 +1121,16 @@ int efa_rdm_cq_open(struct fid_domain *domain, struct fi_cq_attr *attr,
 
 	dlist_init(&cq->ibv_cq_poll_list);
 	cq->need_to_scan_ep_list = false;
+	
 	ret = ofi_cq_init(&efa_prov, domain, attr, &cq->efa_cq.util_cq,
 			  &efa_rdm_cq_progress, context);
 
 	if (ret)
 		goto free;
 
+	cq->efa_cq.wait_obj = attr->wait_obj;
+	cq->efa_cq.wait_cond = attr->wait_cond;
+
 	ret = efa_cq_open_ibv_cq(
 		attr, efa_domain->device->ibv_ctx, &cq->efa_cq.ibv_cq,
 		&efa_cq_init_attr);
@@ -959,6 +1144,7 @@ int efa_rdm_cq_open(struct fid_domain *domain, struct fi_cq_attr *attr,
 	/* open shm cq as peer cq */
 	if (efa_domain->shm_domain) {
 		memcpy(&shm_cq_attr, attr, sizeof(*attr));
+		shm_cq_attr.wait_obj = FI_WAIT_NONE;
 		/* Bind ep with shm provider's cq */
 		shm_cq_attr.flags |= FI_PEER;
 		peer_cq_context.size = sizeof(peer_cq_context);

prov/efa/src/rdm/efa_rdm_cq.h

@@ -17,6 +17,8 @@ struct efa_rdm_cq {
 int efa_rdm_cq_open(struct fid_domain *domain, struct fi_cq_attr *attr,
 		    struct fid_cq **cq_fid, void *context);
 
+int efa_rdm_cq_trywait(struct efa_rdm_cq *cq);
+
 void efa_rdm_cq_poll_ibv_cq_closing_ep(struct efa_ibv_cq *ibv_cq, struct efa_rdm_ep *closing_ep);
 int efa_rdm_cq_poll_ibv_cq(ssize_t cqe_to_process, struct efa_ibv_cq *ibv_cq);
 

prov/efa/src/rdm/efa_rdm_ope.c

@@ -920,6 +920,9 @@ void efa_rdm_rxe_report_completion(struct efa_rdm_ope *rxe)
 			return;
 		}
 
+		if (rx_cq->wait)
+			rx_cq->wait->signal(rx_cq->wait);
+
 		rxe->fi_flags |= EFA_RDM_TXE_NO_COMPLETION;
 	}
 
@@ -1020,6 +1023,9 @@ void efa_rdm_txe_report_completion(struct efa_rdm_ope *txe)
 			efa_rdm_txe_handle_error(txe, -ret, FI_EFA_ERR_WRITE_SEND_COMP);
 			return;
 		}
+
+		if (tx_cq->wait)
+			tx_cq->wait->signal(tx_cq->wait);
 	}
 
 	efa_cntr_report_tx_completion(&txe->ep->base_ep.util_ep, txe->cq_entry.flags);