Skip to content

prov/efa: Add support for blocking CQ read operations for RDM #11568

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
+341 −29
Open

prov/efa: Add support for blocking CQ read operations for RDM #11568

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
54211fb
core: Add `fi_tostr()` support for wait conditions and objects
darrylabbate Oct 29, 2025
8d02d10
fabtests: Drive progress for blocking completion methods
darrylabbate Nov 11, 2025
d638fb5
prov/efa: Only set CQ FID after successful SHM CQ init
darrylabbate Nov 10, 2025
b7bf24d
prov/efa: Add support for blocking CQ read in RDM
darrylabbate Oct 31, 2025
79cc073
prov/efa: Add unit tests for RDM CQ sread
darrylabbate Oct 31, 2025
8b8e568
fabtests/efa: Enable sread tests for `efa` fabric
darrylabbate Nov 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 55 additions & 3 deletions fabtests/common/shared.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
#include <string.h>
#include <unistd.h>
#include <sched.h>
#include <stdbool.h>
#include <sys/wait.h>
#include <sys/types.h>
#include <sys/socket.h>
Expand Down Expand Up @@ -78,6 +79,8 @@ struct ft_context *tx_ctx_arr = NULL, *rx_ctx_arr = NULL;
uint64_t remote_cq_data = 0;

uint64_t tx_seq, rx_seq, tx_cq_cntr, rx_cq_cntr;

static bool ft_fdwait_need_progress;
int (*ft_mr_alloc_func)(void);
uint64_t ft_tag = 0;
int ft_parent_proc = 0;
Expand Down Expand Up @@ -108,6 +111,8 @@ struct timespec start, end;
int listen_sock = -1;
int sock = -1;
int oob_sock = -1;
#define FT_FDWAIT_PROGRESS_INTERVAL_MS 10

bool allow_rx_cq_data = true;

struct fi_av_attr av_attr = {
Expand Down Expand Up @@ -768,6 +773,16 @@ int ft_open_fabric_res(void)
return ft_open_domain_res();
}

static inline void ft_update_fdwait_progress_requirement(struct fi_info *info)
{
bool manual_progress = info && info->domain_attr &&
(info->domain_attr->data_progress == FI_PROGRESS_MANUAL ||
info->domain_attr->control_progress == FI_PROGRESS_MANUAL);

ft_fdwait_need_progress = manual_progress &&
(opts.comp_method == FT_COMP_WAIT_FD);
}

int ft_alloc_ep_res(struct fi_info *fi, struct fid_cq **new_txcq,
struct fid_cq **new_rxcq, struct fid_cntr **new_txcntr,
struct fid_cntr **new_rxcntr,
Expand All @@ -776,6 +791,8 @@ int ft_alloc_ep_res(struct fi_info *fi, struct fid_cq **new_txcq,
{
int ret;

ft_update_fdwait_progress_requirement(fi);

if (cq_attr.format == FI_CQ_FORMAT_UNSPEC) {
if (fi->caps & FI_TAGGED)
cq_attr.format = FI_CQ_FORMAT_TAGGED;
Expand Down Expand Up @@ -1883,6 +1900,20 @@ static void ft_cleanup_mr_array(struct ft_context *ctx_arr, char **mr_bufs)
}
}

static inline int ft_fdwait_poll_timeout(int remaining)
{
int interval;

if (!ft_fdwait_need_progress)
return remaining;

if (remaining < 0)
return FT_FDWAIT_PROGRESS_INTERVAL_MS;

interval = MIN(remaining, FT_FDWAIT_PROGRESS_INTERVAL_MS);
return interval > 0 ? interval : FT_FDWAIT_PROGRESS_INTERVAL_MS;
Comment on lines +1910 to +1914
Copy link
Contributor

@j-xiong j-xiong Nov 22, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can be simplified as:

	if (remaining <= 0)
		return FT_FDWAIT_PROGRESS_INTERVAL_MS;

	return MIN(remaining, FT_FDWAIT_INTERVAL_MS);

And the interval variable can be removed,

}

void ft_close_fids(void)
{
FT_CLOSE_FID(mc);
Expand Down Expand Up @@ -2749,6 +2780,7 @@ static int ft_wait_for_comp(struct fid_cq *cq, uint64_t *cur,

while (total - *cur > 0) {
ret = fi_cq_sread(cq, &comp, 1, NULL, timeout);

if (ret > 0) {
if (!ft_tag_is_valid(cq, &comp, tag ? tag : rx_cq_cntr))
return -FI_EOTHER;
Expand All @@ -2770,17 +2802,36 @@ static int ft_fdwait_for_comp(struct fid_cq *cq, uint64_t *cur,
{
struct fi_cq_err_entry comp;
struct fid *fids[1];
int fd, ret;
int fd, ret = 0;
int remaining = timeout;

fd = cq == txcq ? tx_fd : rx_fd;
fids[0] = &cq->fid;

while (total - *cur > 0) {
ret = fi_trywait(fabric, fids, 1);
if (ret == FI_SUCCESS) {
ret = ft_poll_fd(fd, timeout);
int wait_timeout = ft_fdwait_need_progress
? ft_fdwait_poll_timeout(remaining)
: remaining;
Comment on lines +2814 to +2816
Copy link
Contributor

@j-xiong j-xiong Nov 22, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the same condition is checked inside ft_fdwait_poll_timeout(). can eliminate this variable and use ft_fdwait_poll_timeout(remaining) directly.


if (ft_fdwait_need_progress)
(void) fi_cq_read(cq, NULL, 0);

ret = ft_poll_fd(fd, wait_timeout);
if (ret && ret != -FI_EAGAIN)
return ret;

if (ret == -FI_EAGAIN) {
if (remaining >= 0) {
remaining -= wait_timeout;
if (remaining <= 0)
return -FI_EAGAIN;
}
continue;
}
} else if (ret && ret != -FI_EAGAIN) {
return ret;
}

ret = fi_cq_read(cq, &comp, 1);
Expand All @@ -2791,9 +2842,10 @@ static int ft_fdwait_for_comp(struct fid_cq *cq, uint64_t *cur,
} else if (ret < 0 && ret != -FI_EAGAIN) {
return ret;
}
ret = 0;
Copy link
Contributor

@j-xiong j-xiong Nov 22, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This becomes unnecessary as well.

}

return 0;
return ret;
Comment on lines -2796 to +2848
Copy link
Contributor

@j-xiong j-xiong Nov 22, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is no error path reaching here, can just return 0.

}

int ft_read_cq(struct fid_cq *cq, uint64_t *cur,
Expand Down