From 2ca3b5eec2ea20cf7c3db89acde885d3a2f91ee9 Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Thu, 10 Sep 2026 15:50:11 +0800 Subject: [PATCH 1/8] AUTHORS: add Guangguan Wang Signed-off-by: Guangguan Wang --- AUTHORS | 1 + 1 file changed, 1 insertion(+) diff --git a/AUTHORS b/AUTHORS index 1f7b2218fb9..b048c440ce0 100644 --- a/AUTHORS +++ b/AUTHORS @@ -42,6 +42,7 @@ Gilbert Lee Gilles Gouaillardet Gonzalo Brito Gadeschi Graham Lopez +Guangguan Wang Guy Ealey Morag Guy Shattah Hannes Hansen From 0f4f496c6201d959032af70293f81a40e89fac53 Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Wed, 9 Sep 2026 20:26:46 +0800 Subject: [PATCH 2/8] UCT/IB/RC: Add RC interface initialization attributes The RC interface constructor takes uct_ib_iface_init_attr_t, so an RC transport cannot pass an attribute which only the RC layer understands. Add uct_rc_iface_init_attr_t, which wraps it, and pass that instead. No attribute is added yet, so the behavior is unchanged. Signed-off-by: Guangguan Wang --- src/uct/ib/mlx5/dc/dc_mlx5.c | 22 ++++++++--------- src/uct/ib/mlx5/gdaki/gdaki.c | 11 +++++---- src/uct/ib/mlx5/gga/gga_mlx5.c | 20 ++++++++-------- src/uct/ib/mlx5/rc/rc_mlx5_common.h | 2 +- src/uct/ib/mlx5/rc/rc_mlx5_iface.c | 36 ++++++++++++++-------------- src/uct/ib/rc/base/rc_iface.c | 14 ++++++----- src/uct/ib/rc/base/rc_iface.h | 7 +++++- src/uct/ib/rc/verbs/rc_verbs_iface.c | 24 ++++++++++--------- 8 files changed, 73 insertions(+), 63 deletions(-) diff --git a/src/uct/ib/mlx5/dc/dc_mlx5.c b/src/uct/ib/mlx5/dc/dc_mlx5.c index eadf92ccfea..aef160e4ff7 100644 --- a/src/uct/ib/mlx5/dc/dc_mlx5.c +++ b/src/uct/ib/mlx5/dc/dc_mlx5.c @@ -1667,7 +1667,7 @@ static UCS_CLASS_INIT_FUNC(uct_dc_mlx5_iface_t, uct_md_h tl_md, uct_worker_h wor uct_dc_mlx5_iface_config_t); uct_ib_mlx5_md_t *md = ucs_derived_of(tl_md, uct_ib_mlx5_md_t); - uct_ib_iface_init_attr_t init_attr = {}; + uct_rc_iface_init_attr_t init_attr = {}; unsigned tx_queue_len = config->super.super.tx.queue_len; size_t sq_length; ucs_status_t status; @@ -1697,20 +1697,20 @@ static UCS_CLASS_INIT_FUNC(uct_dc_mlx5_iface_t, uct_md_h tl_md, uct_worker_h wor UCT_DC_MLX5_HW_DCI_INDEX : -1; - init_attr.qp_type = UCT_IB_QPT_DCI; - init_attr.flags = UCT_IB_TX_OPS_PER_PATH; - init_attr.fc_req_size = sizeof(uct_dc_fc_request_t); - init_attr.max_rd_atomic = md->max_rd_atomic_dc; - init_attr.tx_moderation = 0; /* disable tx moderation for dcs */ - init_attr.dev_name = params->mode.device.dev_name; + init_attr.super.qp_type = UCT_IB_QPT_DCI; + init_attr.super.flags = UCT_IB_TX_OPS_PER_PATH; + init_attr.super.fc_req_size = sizeof(uct_dc_fc_request_t); + init_attr.super.max_rd_atomic = md->max_rd_atomic_dc; + init_attr.super.tx_moderation = 0; /* disable tx moderation for dcs */ + init_attr.super.dev_name = params->mode.device.dev_name; if (md->flags & UCT_IB_MLX5_MD_FLAG_DC_TM) { - init_attr.flags |= UCT_IB_TM_SUPPORTED; + init_attr.super.flags |= UCT_IB_TM_SUPPORTED; } if ((md->dp_ordering_cap_devx.dc == UCT_IB_MLX5_DP_ORDERING_OOO_ALL) || md->ddp_support_dv.dc) { - init_attr.flags |= UCT_IB_DDP_SUPPORTED; + init_attr.super.flags |= UCT_IB_DDP_SUPPORTED; } status = uct_dc_mlx5_calc_sq_length(md, tx_queue_len, &sq_length); @@ -1718,7 +1718,7 @@ static UCS_CLASS_INIT_FUNC(uct_dc_mlx5_iface_t, uct_md_h tl_md, uct_worker_h wor return status; } - init_attr.cq_len[UCT_IB_DIR_TX] = sq_length * self->tx.ndci; + init_attr.super.cq_len[UCT_IB_DIR_TX] = sq_length * self->tx.ndci; status = uct_rc_mlx5_dp_ordering_ooo_init(md, &self->super, md->dp_ordering_cap_devx.dc, @@ -1734,7 +1734,7 @@ static UCS_CLASS_INIT_FUNC(uct_dc_mlx5_iface_t, uct_md_h tl_md, uct_worker_h wor tl_md, worker, params, &config->super, &config->rc_mlx5_common, &init_attr); - tx_cq_size = uct_ib_cq_size(&self->super.super.super, &init_attr, + tx_cq_size = uct_ib_cq_size(&self->super.super.super, &init_attr.super, UCT_IB_DIR_TX); /* driver will round up num cqes to pow of 2 if needed */ diff --git a/src/uct/ib/mlx5/gdaki/gdaki.c b/src/uct/ib/mlx5/gdaki/gdaki.c index 1cc703cd2af..53de4b857f1 100644 --- a/src/uct/ib/mlx5/gdaki/gdaki.c +++ b/src/uct/ib/mlx5/gdaki/gdaki.c @@ -1153,7 +1153,7 @@ static UCS_CLASS_INIT_FUNC(uct_rc_gdaki_iface_t, uct_md_h tl_md, uct_rc_gdaki_iface_config_t *config = ucs_derived_of(tl_config, uct_rc_gdaki_iface_config_t); uct_ib_mlx5_md_t *md = ucs_derived_of(tl_md, uct_ib_mlx5_md_t); - uct_ib_iface_init_attr_t init_attr = {}; + uct_rc_iface_init_attr_t init_attr = {}; uct_md_mem_reg_params_t reg_params = {}; UCS_STRING_BUFFER_ONSTACK(strb, 64); char *gpu_name, *ib_name; @@ -1183,10 +1183,11 @@ static UCS_CLASS_INIT_FUNC(uct_rc_gdaki_iface_t, uct_md_h tl_md, gpu_name = ucs_string_buffer_next_token(&strb, NULL, "-"); ib_name = ucs_string_buffer_next_token(&strb, gpu_name, "-"); - init_attr.seg_size = config->super.super.seg_size; - init_attr.qp_type = IBV_QPT_RC; - init_attr.dev_name = ib_name; - init_attr.max_rd_atomic = IBV_DEV_ATTR(&md->super.dev, max_qp_rd_atom); + init_attr.super.seg_size = config->super.super.seg_size; + init_attr.super.qp_type = IBV_QPT_RC; + init_attr.super.dev_name = ib_name; + init_attr.super.max_rd_atomic = IBV_DEV_ATTR(&md->super.dev, + max_qp_rd_atom); UCS_CLASS_CALL_SUPER_INIT(uct_rc_mlx5_iface_common_t, &uct_rc_gdaki_iface_tl_ops, diff --git a/src/uct/ib/mlx5/gga/gga_mlx5.c b/src/uct/ib/mlx5/gga/gga_mlx5.c index a804fbab21c..8c816b65da8 100644 --- a/src/uct/ib/mlx5/gga/gga_mlx5.c +++ b/src/uct/ib/mlx5/gga/gga_mlx5.c @@ -773,19 +773,19 @@ static UCS_CLASS_INIT_FUNC(uct_gga_mlx5_iface_t, uct_gga_mlx5_iface_config_t *config = ucs_derived_of(tl_config, uct_gga_mlx5_iface_config_t); uct_ib_mlx5_md_t *md = ucs_derived_of(tl_md, uct_ib_mlx5_md_t); - uct_ib_iface_init_attr_t init_attr = {}; + uct_rc_iface_init_attr_t init_attr = {}; ucs_status_t status; uct_ib_mlx5_dp_ordering_t dp_ordering; - init_attr.xport_hdr_len = UCT_IB_RETH_LEN; - init_attr.qp_type = IBV_QPT_RC; - init_attr.cq_len[UCT_IB_DIR_TX] = config->super.tx_cq_len; - init_attr.max_rd_atomic = IBV_DEV_ATTR(&md->super.dev, - max_qp_rd_atom); - init_attr.tx_moderation = config->super.tx_cq_moderation; - init_attr.dev_name = params->mode.device.dev_name; - dp_ordering = ucs_min(md->dp_ordering_cap_devx.rc, - UCT_IB_MLX5_DP_ORDERING_OOO_RW); + init_attr.super.xport_hdr_len = UCT_IB_RETH_LEN; + init_attr.super.qp_type = IBV_QPT_RC; + init_attr.super.cq_len[UCT_IB_DIR_TX] = config->super.tx_cq_len; + init_attr.super.max_rd_atomic = IBV_DEV_ATTR(&md->super.dev, + max_qp_rd_atom); + init_attr.super.tx_moderation = config->super.tx_cq_moderation; + init_attr.super.dev_name = params->mode.device.dev_name; + dp_ordering = ucs_min(md->dp_ordering_cap_devx.rc, + UCT_IB_MLX5_DP_ORDERING_OOO_RW); status = uct_rc_mlx5_dp_ordering_ooo_init(md, &self->super, dp_ordering, 0, &config->rc_mlx5_common, "gga"); diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_common.h b/src/uct/ib/mlx5/rc/rc_mlx5_common.h index c6283abdc59..4915377f0bb 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_common.h +++ b/src/uct/ib/mlx5/rc/rc_mlx5_common.h @@ -441,7 +441,7 @@ UCS_CLASS_DECLARE(uct_rc_mlx5_iface_common_t, uct_iface_ops_t*, uct_rc_iface_ops_t*, uct_md_h, uct_worker_h, const uct_iface_params_t*, uct_rc_iface_common_config_t*, uct_rc_mlx5_iface_common_config_t*, - uct_ib_iface_init_attr_t*); + uct_rc_iface_init_attr_t*); #define UCT_RC_MLX5_TM_STAT(_iface, _op) \ diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c index 8f2c00d3466..53814802b85 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c @@ -826,7 +826,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_mlx5_iface_common_t, uct_iface_ops_t *tl_ops, uct_worker_h worker, const uct_iface_params_t *params, uct_rc_iface_common_config_t *rc_config, uct_rc_mlx5_iface_common_config_t *mlx5_config, - uct_ib_iface_init_attr_t *init_attr) + uct_rc_iface_init_attr_t *init_attr) { uct_ib_mlx5_md_t *md = ucs_derived_of(tl_md, uct_ib_mlx5_md_t); uct_ib_device_t *dev; @@ -844,20 +844,20 @@ UCS_CLASS_INIT_FUNC(uct_rc_mlx5_iface_common_t, uct_iface_ops_t *tl_ops, return UCS_ERR_INVALID_PARAM; } - init_attr->flags |= UCT_IB_CQ_IGNORE_OVERRUN; - uct_ib_mlx5_parse_cqe_zipping(md, &mlx5_config->super, init_attr); + init_attr->super.flags |= UCT_IB_CQ_IGNORE_OVERRUN; + uct_ib_mlx5_parse_cqe_zipping(md, &mlx5_config->super, &init_attr->super); status = uct_rc_mlx5_iface_preinit(self, tl_md, rc_config, mlx5_config, - params, init_attr); + params, &init_attr->super); if (status != UCS_OK) { return status; } self->rx.srq.type = UCT_IB_MLX5_OBJ_TYPE_LAST; self->tm.cmd_wq.super.super.type = UCT_IB_MLX5_OBJ_TYPE_LAST; - init_attr->rx_hdr_len = UCT_RC_MLX5_MP_ENABLED(self) ? + init_attr->super.rx_hdr_len = UCT_RC_MLX5_MP_ENABLED(self) ? 0 : sizeof(uct_rc_mlx5_hdr_t); - init_attr->xport_hdr_len = ucs_max(sizeof(uct_rc_hdr_t), + init_attr->super.xport_hdr_len = ucs_max(sizeof(uct_rc_hdr_t), UCT_IB_RETH_LEN); UCS_CLASS_CALL_SUPER_INIT(uct_rc_iface_t, tl_ops, ops, tl_md, worker, @@ -905,7 +905,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_mlx5_iface_common_t, uct_iface_ops_t *tl_ops, status = uct_rc_mlx5_iface_init_fence_flags( self, rc_config, md, dev, - (init_attr->qp_type == IBV_QPT_RC) && + (init_attr->super.qp_type == IBV_QPT_RC) && md->super.relaxed_order_required); if (status != UCS_OK) { goto cleanup_dm; @@ -996,22 +996,22 @@ UCS_CLASS_INIT_FUNC(uct_rc_mlx5_iface_t, uct_rc_mlx5_iface_config_t *config = ucs_derived_of(tl_config, uct_rc_mlx5_iface_config_t); uct_ib_mlx5_md_t *md = ucs_derived_of(tl_md, uct_ib_mlx5_md_t); - uct_ib_iface_init_attr_t init_attr = {}; + uct_rc_iface_init_attr_t init_attr = {}; ucs_status_t status; - init_attr.fc_req_size = sizeof(uct_rc_pending_req_t); - init_attr.flags = IBV_DEVICE_TM_FLAGS(&md->super.dev) ? - UCT_IB_TM_SUPPORTED : 0; - init_attr.cq_len[UCT_IB_DIR_TX] = config->super.tx_cq_len; - init_attr.qp_type = IBV_QPT_RC; - init_attr.max_rd_atomic = IBV_DEV_ATTR(&md->super.dev, - max_qp_rd_atom); - init_attr.tx_moderation = config->super.tx_cq_moderation; - init_attr.dev_name = params->mode.device.dev_name; + init_attr.super.fc_req_size = sizeof(uct_rc_pending_req_t); + init_attr.super.flags = + IBV_DEVICE_TM_FLAGS(&md->super.dev) ? UCT_IB_TM_SUPPORTED : 0; + init_attr.super.cq_len[UCT_IB_DIR_TX] = config->super.tx_cq_len; + init_attr.super.qp_type = IBV_QPT_RC; + init_attr.super.max_rd_atomic = IBV_DEV_ATTR(&md->super.dev, + max_qp_rd_atom); + init_attr.super.tx_moderation = config->super.tx_cq_moderation; + init_attr.super.dev_name = params->mode.device.dev_name; if ((md->dp_ordering_cap_devx.rc == UCT_IB_MLX5_DP_ORDERING_OOO_ALL) || md->ddp_support_dv.rc) { - init_attr.flags |= UCT_IB_DDP_SUPPORTED; + init_attr.super.flags |= UCT_IB_DDP_SUPPORTED; } status = uct_rc_mlx5_dp_ordering_ooo_init(md, &self->super, diff --git a/src/uct/ib/rc/base/rc_iface.c b/src/uct/ib/rc/base/rc_iface.c index d2deba0cd47..ed3df7fdd94 100644 --- a/src/uct/ib/rc/base/rc_iface.c +++ b/src/uct/ib/rc/base/rc_iface.c @@ -572,7 +572,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, uct_rc_iface_ops_t *ops, uct_md_h tl_md, uct_worker_h worker, const uct_iface_params_t *params, const uct_rc_iface_common_config_t *config, - const uct_ib_iface_init_attr_t *init_attr) + const uct_rc_iface_init_attr_t *init_attr) { uct_ib_md_t *md = ucs_derived_of(tl_md, uct_ib_md_t); uct_ib_device_t *dev = &md->dev; @@ -582,9 +582,11 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, ucs_mpool_params_t mp_params; UCS_CLASS_CALL_SUPER_INIT(uct_ib_iface_t, tl_ops, &ops->super, tl_md, - worker, params, &config->super, init_attr); + worker, params, &config->super, + &init_attr->super); - tx_cq_size = uct_ib_cq_size(&self->super, init_attr, + tx_cq_size = uct_ib_cq_size(&self->super, + &init_attr->super, UCT_IB_DIR_TX); /* Prevent title CQE overwriting */ self->tx.cq_available = uct_rc_iface_tx_cq_capacity(tx_cq_size); @@ -593,7 +595,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, self->config.tx_qp_len = config->super.tx.queue_len; self->config.tx_min_sge = config->super.tx.min_sge; self->config.tx_min_inline = config->super.tx.min_inline; - self->config.tx_moderation = init_attr->tx_moderation; + self->config.tx_moderation = init_attr->super.tx_moderation; self->config.tx_poll_always = config->tx.poll_always; self->config.tx_cq_len = tx_cq_size; self->config.min_rnr_timer = uct_ib_to_rnr_fabric_time(config->tx.rnr_timeout); @@ -620,7 +622,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, self->config.ece = config->ece; - status = uct_rc_iface_init_max_rd_atomic(self, config, init_attr); + status = uct_rc_iface_init_max_rd_atomic(self, config, &init_attr->super); if (status != UCS_OK) { goto err; } @@ -703,7 +705,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, /* Create mempool for pending requests */ ucs_mpool_params_reset(&mp_params); - mp_params.elem_size = ucs_max(init_attr->fc_req_size, + mp_params.elem_size = ucs_max(init_attr->super.fc_req_size, sizeof(uct_rc_pending_req_t)); mp_params.alignment = 1; mp_params.elems_per_chunk = 128; diff --git a/src/uct/ib/rc/base/rc_iface.h b/src/uct/ib/rc/base/rc_iface.h index 737fc4de75d..f4d3a6d9adc 100644 --- a/src/uct/ib/rc/base/rc_iface.h +++ b/src/uct/ib/rc/base/rc_iface.h @@ -250,6 +250,11 @@ typedef struct uct_rc_srq { } uct_rc_srq_t; +typedef struct uct_rc_iface_init_attr { + uct_ib_iface_init_attr_t super; +} uct_rc_iface_init_attr_t; + + struct uct_rc_iface { uct_ib_iface_t super; @@ -328,7 +333,7 @@ struct uct_rc_iface { UCS_CLASS_DECLARE(uct_rc_iface_t, uct_iface_ops_t*, uct_rc_iface_ops_t*, uct_md_h, uct_worker_h, const uct_iface_params_t*, const uct_rc_iface_common_config_t*, - const uct_ib_iface_init_attr_t*); + const uct_rc_iface_init_attr_t*); struct uct_rc_iface_send_op { diff --git a/src/uct/ib/rc/verbs/rc_verbs_iface.c b/src/uct/ib/rc/verbs/rc_verbs_iface.c index f548cdd3968..e6943f5f97a 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_iface.c +++ b/src/uct/ib/rc/verbs/rc_verbs_iface.c @@ -258,7 +258,7 @@ static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, ucs_derived_of(tl_config, uct_rc_verbs_iface_config_t); uct_ib_iface_config_t *ib_config = &config->super.super.super; uct_ib_md_t *ib_md = ucs_derived_of(tl_md, uct_ib_md_t); - uct_ib_iface_init_attr_t init_attr = {}; + uct_rc_iface_init_attr_t init_attr = {}; uct_ib_qp_attr_t attr = {}; const char *dev_name; ucs_status_t status; @@ -269,16 +269,18 @@ static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, return UCS_ERR_UNSUPPORTED; } - init_attr.fc_req_size = sizeof(uct_rc_pending_req_t); - init_attr.rx_hdr_len = sizeof(uct_rc_hdr_t); - init_attr.qp_type = IBV_QPT_RC; - init_attr.cq_len[UCT_IB_DIR_RX] = ib_config->rx.queue_len; - init_attr.cq_len[UCT_IB_DIR_TX] = config->super.tx_cq_len; - init_attr.seg_size = ib_config->seg_size; - init_attr.xport_hdr_len = ucs_max(sizeof(uct_rc_hdr_t), UCT_IB_RETH_LEN); - init_attr.max_rd_atomic = IBV_DEV_ATTR(&ib_md->dev, max_qp_rd_atom); - init_attr.tx_moderation = config->super.tx_cq_moderation; - init_attr.dev_name = params->mode.device.dev_name; + init_attr.super.fc_req_size = sizeof(uct_rc_pending_req_t); + init_attr.super.rx_hdr_len = sizeof(uct_rc_hdr_t); + init_attr.super.qp_type = IBV_QPT_RC; + init_attr.super.cq_len[UCT_IB_DIR_RX] = ib_config->rx.queue_len; + init_attr.super.cq_len[UCT_IB_DIR_TX] = config->super.tx_cq_len; + init_attr.super.seg_size = ib_config->seg_size; + init_attr.super.xport_hdr_len = ucs_max(sizeof(uct_rc_hdr_t), + UCT_IB_RETH_LEN); + init_attr.super.max_rd_atomic = IBV_DEV_ATTR(&ib_md->dev, + max_qp_rd_atom); + init_attr.super.tx_moderation = config->super.tx_cq_moderation; + init_attr.super.dev_name = params->mode.device.dev_name; UCS_CLASS_CALL_SUPER_INIT(uct_rc_iface_t, &uct_rc_verbs_iface_tl_ops, &uct_rc_verbs_iface_ops, tl_md, worker, params, From abaaa2d52486e9e51e2a2ec65e7c0297e068fbe7 Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Wed, 9 Sep 2026 20:26:58 +0800 Subject: [PATCH 3/8] UCT/IB/RC: Add receive queue length to QP creation uct_rc_iface_fill_attr() always sets max_recv_wr to zero, because every RC transport attaches its QPs to an SRQ. Take the length as an argument, so that a QP can be created without an SRQ. All the callers pass zero, so the behavior is unchanged. Signed-off-by: Guangguan Wang --- src/uct/ib/mlx5/gdaki/gdaki.c | 6 +++--- src/uct/ib/mlx5/rc/rc_mlx5_common.c | 5 +++-- src/uct/ib/rc/base/rc_iface.c | 9 +++++---- src/uct/ib/rc/base/rc_iface.h | 3 ++- src/uct/ib/rc/verbs/rc_verbs_ep.c | 3 ++- src/uct/ib/rc/verbs/rc_verbs_iface.c | 2 +- test/gtest/uct/ib/test_ib_event.cc | 2 +- 7 files changed, 17 insertions(+), 13 deletions(-) diff --git a/src/uct/ib/mlx5/gdaki/gdaki.c b/src/uct/ib/mlx5/gdaki/gdaki.c index 53de4b857f1..e95f338eed8 100644 --- a/src/uct/ib/mlx5/gdaki/gdaki.c +++ b/src/uct/ib/mlx5/gdaki/gdaki.c @@ -483,7 +483,7 @@ uct_rc_gdaki_init_channel_chunk(uct_rc_gdaki_iface_t *iface, uct_ib_mlx5_cq_calc_sizes(&iface->super.super.super, UCT_IB_DIR_TX, &init_attr, 0, &cq_attr); uct_rc_iface_fill_attr(&iface->super.super, &qp_attr.super, - iface->super.super.config.tx_qp_len, NULL); + iface->super.super.config.tx_qp_len, 0, NULL); cq_attr.flags |= UCT_IB_MLX5_CQ_IGNORE_OVERRUN; qp_attr.mmio_mode = UCT_IB_MLX5_MMIO_MODE_DB; @@ -647,7 +647,7 @@ uct_rc_gdaki_iface_init_channel_pool(uct_rc_gdaki_iface_t *iface, } uct_rc_iface_fill_attr(&iface->super.super, &qp_attr.super, - iface->super.super.config.tx_qp_len, NULL); + iface->super.super.config.tx_qp_len, 0, NULL); uct_ib_mlx5_wq_calc_sizes(&qp_attr); priv = ucs_mpool_priv(&iface->channel_pool); @@ -711,7 +711,7 @@ uct_rc_gdaki_ep_init_channels_direct(uct_rc_gdaki_iface_t *iface, ucs_status_t status; uct_rc_iface_fill_attr(&iface->super.super, &qp_attr.super, - iface->super.super.config.tx_qp_len, NULL); + iface->super.super.config.tx_qp_len, 0, NULL); uct_ib_mlx5_wq_calc_sizes(&qp_attr); uct_rc_gdaki_calc_dev_ep_layout(iface->num_channels, qp_attr.len, &dev_ep_size, &pgsz_bitmap); diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_common.c b/src/uct/ib/mlx5/rc/rc_mlx5_common.c index 0b29cba8b95..c2760c23213 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_common.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_common.c @@ -528,12 +528,13 @@ void uct_rc_mlx5_iface_fill_attr(uct_rc_mlx5_iface_common_t *iface, { switch (srq->type) { case UCT_IB_MLX5_OBJ_TYPE_VERBS: - uct_rc_iface_fill_attr(&iface->super, &qp_attr->super, max_send_wr, + uct_rc_iface_fill_attr(&iface->super, &qp_attr->super, max_send_wr, 0, srq->verbs.srq); break; case UCT_IB_MLX5_OBJ_TYPE_DEVX: case UCT_IB_MLX5_OBJ_TYPE_NULL: - uct_rc_iface_fill_attr(&iface->super, &qp_attr->super, max_send_wr, NULL); + uct_rc_iface_fill_attr(&iface->super, &qp_attr->super, max_send_wr, 0, + NULL); break; case UCT_IB_MLX5_OBJ_TYPE_LAST: break; diff --git a/src/uct/ib/rc/base/rc_iface.c b/src/uct/ib/rc/base/rc_iface.c index ed3df7fdd94..83fb7f47c89 100644 --- a/src/uct/ib/rc/base/rc_iface.c +++ b/src/uct/ib/rc/base/rc_iface.c @@ -812,11 +812,12 @@ static UCS_CLASS_CLEANUP_FUNC(uct_rc_iface_t) UCS_CLASS_DEFINE(uct_rc_iface_t, uct_ib_iface_t); void uct_rc_iface_fill_attr(uct_rc_iface_t *iface, uct_ib_qp_attr_t *attr, - unsigned max_send_wr, struct ibv_srq *srq) + unsigned max_send_wr, unsigned max_recv_wr, + struct ibv_srq *srq) { attr->srq = srq; attr->cap.max_send_wr = max_send_wr; - attr->cap.max_recv_wr = 0; + attr->cap.max_recv_wr = max_recv_wr; attr->cap.max_send_sge = iface->config.tx_min_sge; attr->cap.max_recv_sge = 1; attr->cap.max_inline_data = iface->config.tx_min_inline; @@ -828,9 +829,9 @@ void uct_rc_iface_fill_attr(uct_rc_iface_t *iface, uct_ib_qp_attr_t *attr, ucs_status_t uct_rc_iface_qp_create(uct_rc_iface_t *iface, struct ibv_qp **qp_p, uct_ib_qp_attr_t *attr, unsigned max_send_wr, - struct ibv_srq *srq) + unsigned max_recv_wr, struct ibv_srq *srq) { - uct_rc_iface_fill_attr(iface, attr, max_send_wr, srq); + uct_rc_iface_fill_attr(iface, attr, max_send_wr, max_recv_wr, srq); return uct_ib_iface_create_qp(&iface->super, attr, qp_p); } diff --git a/src/uct/ib/rc/base/rc_iface.h b/src/uct/ib/rc/base/rc_iface.h index f4d3a6d9adc..b40a17e732f 100644 --- a/src/uct/ib/rc/base/rc_iface.h +++ b/src/uct/ib/rc/base/rc_iface.h @@ -413,11 +413,12 @@ unsigned uct_rc_iface_qp_cleanup_progress(void *arg); */ ucs_status_t uct_rc_iface_qp_create(uct_rc_iface_t *iface, struct ibv_qp **qp_p, uct_ib_qp_attr_t *attr, unsigned max_send_wr, - struct ibv_srq *srq); + unsigned max_recv_wr, struct ibv_srq *srq); void uct_rc_iface_fill_attr(uct_rc_iface_t *iface, uct_ib_qp_attr_t *qp_init_attr, unsigned max_send_wr, + unsigned max_recv_wr, struct ibv_srq *srq); ucs_status_t uct_rc_iface_qp_init(uct_rc_iface_t *iface, struct ibv_qp *qp); diff --git a/src/uct/ib/rc/verbs/rc_verbs_ep.c b/src/uct/ib/rc/verbs/rc_verbs_ep.c index 77d35212295..52d2112e6a0 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_ep.c +++ b/src/uct/ib/rc/verbs/rc_verbs_ep.c @@ -696,7 +696,8 @@ UCS_CLASS_INIT_FUNC(uct_rc_verbs_ep_t, const uct_ep_params_t *params) ucs_status_t status; status = uct_rc_iface_qp_create(&iface->super, &self->qp, &attr, - iface->super.config.tx_qp_len, iface->srq); + iface->super.config.tx_qp_len, 0, + iface->srq); if (status != UCS_OK) { goto err; } diff --git a/src/uct/ib/rc/verbs/rc_verbs_iface.c b/src/uct/ib/rc/verbs/rc_verbs_iface.c index e6943f5f97a..8dccafa69eb 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_iface.c +++ b/src/uct/ib/rc/verbs/rc_verbs_iface.c @@ -351,7 +351,7 @@ static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, /* Create a dummy QP in order to find out max_inline */ status = uct_rc_iface_qp_create(&self->super, &qp, &attr, - self->super.config.tx_qp_len, + self->super.config.tx_qp_len, 0, self->srq); if (status != UCS_OK) { goto err_common_cleanup; diff --git a/test/gtest/uct/ib/test_ib_event.cc b/test/gtest/uct/ib/test_ib_event.cc index 91bab78af67..049629268cb 100644 --- a/test/gtest/uct/ib/test_ib_event.cc +++ b/test/gtest/uct/ib/test_ib_event.cc @@ -387,7 +387,7 @@ class uct_qp_test_event : public uct_test_event_base { ucs_status_t status; status = uct_rc_iface_qp_create(&m_iface->super, &m_ibqp, &attr, - m_iface->super.config.tx_qp_len, + m_iface->super.config.tx_qp_len, 0, m_iface->srq); ASSERT_UCS_OK(status); From ed613fc0436c748e14ae2f9339fe9b1a143b755d Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Wed, 9 Sep 2026 20:27:04 +0800 Subject: [PATCH 4/8] UCT/IB: Add uct_ib_device_has_srq() The device port check finds out whether a device supports SRQ by comparing IBV_DEV_ATTR(dev, max_srq) with zero. Wrap the comparison in a helper, so that the transports can perform the same check. Signed-off-by: Guangguan Wang --- src/uct/ib/base/ib_device.c | 2 +- src/uct/ib/base/ib_device.h | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/src/uct/ib/base/ib_device.c b/src/uct/ib/base/ib_device.c index 94b89dc73d9..544fe245d07 100644 --- a/src/uct/ib/base/ib_device.c +++ b/src/uct/ib/base/ib_device.c @@ -801,7 +801,7 @@ uct_ib_device_port_check(uct_ib_device_t *dev, uint8_t port_num, unsigned flags) } if (flags & UCT_IB_DEVICE_FLAG_SRQ) { - if (IBV_DEV_ATTR(dev, max_srq) == 0) { + if (!uct_ib_device_has_srq(dev)) { ucs_trace("%s:%d does not support SRQ", uct_ib_device_name(dev), port_num); return UCS_ERR_UNSUPPORTED; diff --git a/src/uct/ib/base/ib_device.h b/src/uct/ib/base/ib_device.h index 88dc6c13be2..c8dacd3e755 100644 --- a/src/uct/ib/base/ib_device.h +++ b/src/uct/ib/base/ib_device.h @@ -440,6 +440,11 @@ static inline int uct_ib_device_has_pci_atomics(uct_ib_device_t *dev) (sizeof(uint32_t) | sizeof(uint64_t))); } +static inline int uct_ib_device_has_srq(uct_ib_device_t *dev) +{ + return IBV_DEV_ATTR(dev, max_srq) != 0; +} + const char *uct_ib_roce_version_str(uct_ib_roce_version_t roce_ver); const char *uct_ib_gid_str(const union ibv_gid *gid, char *str, size_t max_size); From dbedcbf0dbd65b3f5452af7d34413bc45789f25d Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Wed, 9 Sep 2026 20:27:09 +0800 Subject: [PATCH 5/8] UCT/IB/RC: Remove an endpoint from ep_list in uct_rc_ep_t cleanup An endpoint is added to iface->ep_list when uct_rc_ep_t is initialized, but it is removed outside of the uct_rc_ep_t cleanup: by rc_verbs and rc_mlx5 through uct_rc_ep_cleanup_qp(), and by gga_mlx5 on its own. If a transport fails to initialize an endpoint, it is freed while it is still on ep_list, and uct_rc_iface_flush() then calls uct_ep_flush() on freed memory. Remove the endpoint in the uct_rc_ep_t cleanup, which runs both when the endpoint is destroyed and when a transport fails to initialize it. Signed-off-by: Guangguan Wang --- src/uct/ib/mlx5/gga/gga_mlx5.c | 1 - src/uct/ib/rc/base/rc_ep.c | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/src/uct/ib/mlx5/gga/gga_mlx5.c b/src/uct/ib/mlx5/gga/gga_mlx5.c index 8c816b65da8..c2faee30ce6 100644 --- a/src/uct/ib/mlx5/gga/gga_mlx5.c +++ b/src/uct/ib/mlx5/gga/gga_mlx5.c @@ -470,7 +470,6 @@ static UCS_CLASS_CLEANUP_FUNC(uct_gga_mlx5_ep_t) uct_rc_iface_remove_qp(&iface->super, self->super.tx.wq.super.qp_num); uct_ib_mlx5_destroy_qp(md, &self->super.tx.wq.super); uct_ib_mlx5_qp_mmio_cleanup(&self->super.tx.wq.super, self->super.tx.wq.reg); - ucs_list_del(&self->super.super.list); uct_rc_iface_add_cq_credits(&iface->super, outstanding - wqe_count); } diff --git a/src/uct/ib/rc/base/rc_ep.c b/src/uct/ib/rc/base/rc_ep.c index 92da16369e5..897f8d229f9 100644 --- a/src/uct/ib/rc/base/rc_ep.c +++ b/src/uct/ib/rc/base/rc_ep.c @@ -118,7 +118,6 @@ void uct_rc_ep_cleanup_qp(uct_rc_ep_t *ep, ucs_assertv(cq_credits < (UINT16_MAX / 2), "cq_credits=%d", cq_credits); - ucs_list_del(&ep->list); uct_rc_iface_remove_qp(iface, qp_num); cleanup_ctx->super.cbq = &iface->super.super.worker->super.progress_q; @@ -186,6 +185,7 @@ static UCS_CLASS_CLEANUP_FUNC(uct_rc_ep_t) ucs_debug("destroy rc ep %p", self); + ucs_list_del(&self->list); uct_rc_ep_pending_purge(&self->super.super, uct_rc_ep_pending_purge_warn_cb, self); uct_rc_fc_cleanup(&self->fc); From 6ca83204836672810c85e6321dc7290a4f0659a7 Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Thu, 10 Sep 2026 17:58:28 +0800 Subject: [PATCH 6/8] UCT/IB/RC/VERBS: Release the descriptor of a failed receive completion A receive descriptor is taken from the interface pool for every receive WR posted to the SRQ, but a completion which reports an error returns without releasing it, while the SRQ counts the WR as available and the next post takes another descriptor from the pool. Every failed receive therefore leaks a descriptor. Release the descriptor before skipping the completion. --- src/uct/ib/rc/verbs/rc_verbs_impl.h | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/uct/ib/rc/verbs/rc_verbs_impl.h b/src/uct/ib/rc/verbs/rc_verbs_impl.h index 14f0d24e9bd..acd743dd8f1 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_impl.h +++ b/src/uct/ib/rc/verbs/rc_verbs_impl.h @@ -94,11 +94,12 @@ uct_rc_verbs_iface_poll_rx_common(uct_rc_verbs_iface_t *iface) desc = (uct_ib_iface_recv_desc_t *)(uintptr_t)wc[i].wr_id; hdr = (uct_rc_hdr_t *)uct_ib_iface_recv_desc_hdr(&iface->super.super, desc); if (ucs_unlikely(wc[i].status != IBV_WC_SUCCESS)) { - if (wc[i].status == IBV_WC_REM_ABORT_ERR) { - continue; - } - /* we can get flushed messages during ep destroy */ - if (wc[i].status == IBV_WC_WR_FLUSH_ERR) { + /* A failed receive does not deliver an active message, so return + * its descriptor to the pool. A flushed one arrives during + * endpoint destruction. */ + if ((wc[i].status == IBV_WC_REM_ABORT_ERR) || + (wc[i].status == IBV_WC_WR_FLUSH_ERR)) { + ucs_mpool_put_inline(desc); continue; } UCT_IB_IFACE_VERBS_COMPLETION_FATAL("receive", &iface->super.super, i, wc); From ac4b1fe3aca26cfbb2d7bc51d570266fc458cd41 Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Wed, 9 Sep 2026 20:27:14 +0800 Subject: [PATCH 7/8] UCT/IB/RC/VERBS: Support QPs without SRQ rc_verbs currently rejects devices whose max_srq is zero. Allow these devices and create every QP with RC_VERBS_RX_MAX_WR receive work requests. RC_VERBS_SRQ_ENABLE controls whether SRQ is required, disabled, or selected according to the device capability. For QPs without SRQ, post and refill receive work requests per endpoint and use the QP number in each receive completion to update the corresponding receive queue. Limit the flow control window to RC_VERBS_RX_MAX_WR, and warn when the sum of the receive queue capacities exceeds the RX CQ capacity. IBV_EVENT_QP_LAST_WQE_REACHED is not generated for QPs without SRQ. Keep a destroyed QP until completions for all its posted receive work requests have been polled, then release the QP and restore its RX CQ capacity. Signed-off-by: Guangguan Wang --- src/uct/ib/rc/base/rc_ep.c | 22 +++- src/uct/ib/rc/base/rc_iface.c | 15 ++- src/uct/ib/rc/base/rc_iface.h | 7 ++ src/uct/ib/rc/verbs/rc_verbs.h | 7 ++ src/uct/ib/rc/verbs/rc_verbs_ep.c | 72 ++++++++++-- src/uct/ib/rc/verbs/rc_verbs_iface.c | 160 ++++++++++++++++++++++++--- src/uct/ib/rc/verbs/rc_verbs_impl.h | 114 ++++++++++++++++--- test/apps/test_fuzzy_match.py | 2 +- 8 files changed, 350 insertions(+), 49 deletions(-) diff --git a/src/uct/ib/rc/base/rc_ep.c b/src/uct/ib/rc/base/rc_ep.c index 897f8d229f9..2034498adc5 100644 --- a/src/uct/ib/rc/base/rc_ep.c +++ b/src/uct/ib/rc/base/rc_ep.c @@ -120,17 +120,27 @@ void uct_rc_ep_cleanup_qp(uct_rc_ep_t *ep, uct_rc_iface_remove_qp(iface, qp_num); - cleanup_ctx->super.cbq = &iface->super.super.worker->super.progress_q; - cleanup_ctx->super.cb = uct_rc_iface_qp_cleanup_progress; cleanup_ctx->iface = iface; cleanup_ctx->qp_num = qp_num; cleanup_ctx->cq_credits = cq_credits; ucs_list_add_tail(&iface->qp_gc_list, &cleanup_ctx->list); - status = uct_ib_device_async_event_wait(&md->dev, - IBV_EVENT_QP_LAST_WQE_REACHED, - qp_num, &cleanup_ctx->super); - ucs_assert_always(status == UCS_OK); + /* The LAST_WQE event is generated only for QPs with SRQ, otherwise the + * transport has to trigger the cleanup by itself by calling + * uct_rc_iface_qp_cleanup_progress() directly */ + if (!iface->config.srq_disable) { + cleanup_ctx->super.cbq = &iface->super.super.worker->super.progress_q; + cleanup_ctx->super.cb = uct_rc_iface_qp_cleanup_progress; + status = uct_ib_device_async_event_wait(&md->dev, + IBV_EVENT_QP_LAST_WQE_REACHED, + qp_num, &cleanup_ctx->super); + ucs_assert_always(status == UCS_OK); + } else { + /* Nothing is registered, so mark the waiting context as empty. */ + cleanup_ctx->super.cb = NULL; + cleanup_ctx->super.cbq = NULL; + cleanup_ctx->super.cb_id = UCS_CALLBACKQ_ID_NULL; + } } UCS_CLASS_INIT_FUNC(uct_rc_ep_t, uct_rc_iface_t *iface, uint32_t qp_num, diff --git a/src/uct/ib/rc/base/rc_iface.c b/src/uct/ib/rc/base/rc_iface.c index 83fb7f47c89..01924bdba85 100644 --- a/src/uct/ib/rc/base/rc_iface.c +++ b/src/uct/ib/rc/base/rc_iface.c @@ -577,6 +577,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, uct_ib_md_t *md = ucs_derived_of(tl_md, uct_ib_md_t); uct_ib_device_t *dev = &md->dev; uint32_t max_ib_msg_size; + unsigned fc_max_wnd_size; ucs_status_t status; unsigned tx_cq_size; ucs_mpool_params_t mp_params; @@ -592,6 +593,7 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, self->tx.cq_available = uct_rc_iface_tx_cq_capacity(tx_cq_size); self->rx.srq.available = 0; self->rx.srq.quota = 0; + self->config.srq_disable = init_attr->srq_disable; self->config.tx_qp_len = config->super.tx.queue_len; self->config.tx_min_sge = config->super.tx.min_sge; self->config.tx_min_inline = config->super.tx.min_inline; @@ -722,8 +724,11 @@ UCS_CLASS_INIT_FUNC(uct_rc_iface_t, uct_iface_ops_t *tl_ops, * Then FC window size is the same for all endpoints as well. * TODO: Make wnd size to be a property of the particular interface. * We could distribute it via rc address then.*/ + fc_max_wnd_size = (init_attr->fc_max_wnd_size != 0) ? + init_attr->fc_max_wnd_size : + config->super.rx.queue_len; self->config.fc_wnd_size = ucs_min(config->fc.wnd_size, - config->super.rx.queue_len); + fc_max_wnd_size); self->config.fc_hard_thresh = ucs_max((int)(self->config.fc_wnd_size * config->fc.hard_thresh), 1); } else { @@ -755,9 +760,11 @@ unsigned uct_rc_iface_qp_cleanup_progress(void *arg) uct_rc_iface_t *iface = cleanup_ctx->iface; uct_rc_iface_ops_t *ops; - uct_ib_device_async_event_unregister(uct_ib_iface_device(&iface->super), - IBV_EVENT_QP_LAST_WQE_REACHED, - cleanup_ctx->qp_num); + if (!iface->config.srq_disable) { + uct_ib_device_async_event_unregister(uct_ib_iface_device(&iface->super), + IBV_EVENT_QP_LAST_WQE_REACHED, + cleanup_ctx->qp_num); + } ops = ucs_derived_of(iface->super.ops, uct_rc_iface_ops_t); ops->cleanup_qp(cleanup_ctx); diff --git a/src/uct/ib/rc/base/rc_iface.h b/src/uct/ib/rc/base/rc_iface.h index b40a17e732f..fbc28441023 100644 --- a/src/uct/ib/rc/base/rc_iface.h +++ b/src/uct/ib/rc/base/rc_iface.h @@ -252,6 +252,11 @@ typedef struct uct_rc_srq { typedef struct uct_rc_iface_init_attr { uct_ib_iface_init_attr_t super; + + uint8_t srq_disable; /* create QPs without an SRQ */ + unsigned fc_max_wnd_size; /* maximum flow control window; + * if zero, the receive queue + * length is used */ } uct_rc_iface_init_attr_t; @@ -304,6 +309,8 @@ struct uct_rc_iface { uint16_t fc_wnd_size; uint8_t fc_enabled; + uint8_t srq_disable; + uint8_t min_rnr_timer; uint8_t timeout; uint8_t rnr_retry; diff --git a/src/uct/ib/rc/verbs/rc_verbs.h b/src/uct/ib/rc/verbs/rc_verbs.h index 6a49fb8ffc2..d260b2ced0a 100644 --- a/src/uct/ib/rc/verbs/rc_verbs.h +++ b/src/uct/ib/rc/verbs/rc_verbs.h @@ -62,6 +62,7 @@ typedef struct uct_rc_verbs_ep { uct_rc_ep_t super; uct_rc_verbs_txcnt_t txcnt; uct_ib_fence_info_t fi; + uint16_t rx_available; /* free receive WR slots without SRQ */ struct ibv_qp *qp; } uct_rc_verbs_ep_t; @@ -70,6 +71,7 @@ typedef struct uct_rc_verbs_ep { typedef struct { uct_rc_iface_qp_cleanup_ctx_t super; struct ibv_qp *qp; + uint16_t rx_remaining; /* receive WRs left to drain */ } uct_rc_verbs_iface_qp_cleanup_ctx_t; @@ -80,7 +82,9 @@ typedef struct uct_rc_verbs_iface_config { uct_rc_iface_config_t super; size_t max_am_hdr; unsigned tx_max_wr; + unsigned rx_max_wr; /* receive WRs per QP */ unsigned flush_mode; + ucs_ternary_auto_value_t srq_enable; } uct_rc_verbs_iface_config_t; @@ -96,11 +100,14 @@ typedef struct uct_rc_verbs_iface { uct_rc_am_short_hdr_t am_inl_hdr; ucs_mpool_t short_desc_mp; uct_rc_iface_send_desc_t *fc_desc; /* used when max_inline is zero */ + int rx_cq_available; /* RX CQ space for new EPs */ + uint8_t rx_post_recv_pending; /* retry QP receives */ struct { size_t short_desc_size; size_t max_inline; size_t max_send_sge; unsigned tx_max_wr; + uint16_t rx_max_wr; /* receive WRs per QP without SRQ */ uint8_t flush_by_fc; } config; } uct_rc_verbs_iface_t; diff --git a/src/uct/ib/rc/verbs/rc_verbs_ep.c b/src/uct/ib/rc/verbs/rc_verbs_ep.c index 52d2112e6a0..3029f91037e 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_ep.c +++ b/src/uct/ib/rc/verbs/rc_verbs_ep.c @@ -688,6 +688,28 @@ uct_rc_verbs_ep_connect_to_ep_v2(uct_ep_h tl_ep, return UCS_OK; } +static ucs_status_t uct_rc_verbs_ep_prepost_recvs(uct_rc_verbs_ep_t *ep) +{ + uct_rc_verbs_iface_t *iface = ucs_derived_of(ep->super.super.super.iface, + uct_rc_verbs_iface_t); + unsigned prev_available = ep->rx_available; + + /* Post as many receive WRs as the shared buffer pool can supply. Progress + * retries the remaining WRs after descriptors are returned to the pool. */ + while (ep->rx_available > 0) { + if (uct_rc_verbs_iface_post_recv_common(iface, ep, 1) == 0) { + if (ep->rx_available == prev_available) { + ucs_error("failed to post receives on QP 0x%x", ep->qp->qp_num); + return UCS_ERR_NO_MEMORY; + } + + break; + } + } + + return UCS_OK; +} + UCS_CLASS_INIT_FUNC(uct_rc_verbs_ep_t, const uct_ep_params_t *params) { uct_rc_verbs_iface_t *iface = ucs_derived_of(params->iface, uct_rc_verbs_iface_t); @@ -695,9 +717,19 @@ UCS_CLASS_INIT_FUNC(uct_rc_verbs_ep_t, const uct_ep_params_t *params) uct_ib_qp_attr_t attr = {}; ucs_status_t status; + if (ucs_unlikely(iface->rx_cq_available < iface->config.rx_max_wr)) { + ucs_error("iface %p: cannot create endpoint, RX CQ of %d entries has " + "no room for %u more receive WRs, increase " + UCS_DEFAULT_ENV_PREFIX "RC_VERBS_MAX_NUM_EPS or decrease " + UCS_DEFAULT_ENV_PREFIX "RC_VERBS_RX_MAX_WR", + iface, iface->super.super.cq[UCT_IB_DIR_RX]->cqe, + iface->config.rx_max_wr); + return UCS_ERR_EXCEEDS_LIMIT; + } + status = uct_rc_iface_qp_create(&iface->super, &self->qp, &attr, - iface->super.config.tx_qp_len, 0, - iface->srq); + iface->super.config.tx_qp_len, + iface->config.rx_max_wr, iface->srq); if (status != UCS_OK) { goto err; } @@ -710,11 +742,13 @@ UCS_CLASS_INIT_FUNC(uct_rc_verbs_ep_t, const uct_ep_params_t *params) goto err_destroy_qp; } - status = uct_ib_device_async_event_register(&md->dev, - IBV_EVENT_QP_LAST_WQE_REACHED, - self->qp->qp_num); - if (status != UCS_OK) { - goto err_destroy_qp; + /* LAST_WQE is generated only for QPs with SRQ. */ + if (!iface->super.config.srq_disable) { + status = uct_ib_device_async_event_register( + &md->dev, IBV_EVENT_QP_LAST_WQE_REACHED, self->qp->qp_num); + if (status != UCS_OK) { + goto err_destroy_qp; + } } status = uct_rc_iface_add_qp(&iface->super, &self->super, self->qp->qp_num); @@ -722,11 +756,19 @@ UCS_CLASS_INIT_FUNC(uct_rc_verbs_ep_t, const uct_ep_params_t *params) goto err_event_unreg; } - status = uct_rc_verbs_iface_common_prepost_recvs(iface); + if (iface->super.config.srq_disable) { + self->rx_available = iface->config.rx_max_wr; + status = uct_rc_verbs_ep_prepost_recvs(self); + } else { + self->rx_available = 0; + status = uct_rc_verbs_iface_common_prepost_recvs(iface); + } if (status != UCS_OK) { goto err_remove_qp; } + iface->rx_cq_available -= iface->config.rx_max_wr; + uct_rc_txqp_available_set(&self->super.txqp, iface->config.tx_max_wr); uct_rc_verbs_txcnt_init(&self->txcnt); uct_ib_fence_info_init(&self->fi); @@ -737,9 +779,11 @@ UCS_CLASS_INIT_FUNC(uct_rc_verbs_ep_t, const uct_ep_params_t *params) err_remove_qp: uct_rc_iface_remove_qp(&iface->super, self->qp->qp_num); err_event_unreg: - uct_ib_device_async_event_unregister(&md->dev, - IBV_EVENT_QP_LAST_WQE_REACHED, - self->qp->qp_num); + if (!iface->super.config.srq_disable) { + uct_ib_device_async_event_unregister(&md->dev, + IBV_EVENT_QP_LAST_WQE_REACHED, + self->qp->qp_num); + } err_destroy_qp: uct_ib_destroy_qp(self->qp); err: @@ -761,6 +805,12 @@ UCS_CLASS_CLEANUP_FUNC(uct_rc_verbs_ep_t) cleanup_ctx = ucs_malloc(sizeof(*cleanup_ctx), "verbs_qp_cleanup_ctx"); ucs_assert_always(cleanup_ctx != NULL); cleanup_ctx->qp = self->qp; + ucs_assertv(self->rx_available <= iface->config.rx_max_wr, + "rx_available=%u rx_max_wr=%u", self->rx_available, + iface->config.rx_max_wr); + /* Without an SRQ, defer QP destruction until every posted receive WR has + * generated a completion and its descriptor has been returned. */ + cleanup_ctx->rx_remaining = iface->config.rx_max_wr - self->rx_available; ucs_assert(UCS_CIRCULAR_COMPARE16(self->txcnt.pi, >=, self->txcnt.ci)); uct_rc_ep_cleanup_qp(&self->super, &cleanup_ctx->super, self->qp->qp_num, self->txcnt.pi - self->txcnt.ci); diff --git a/src/uct/ib/rc/verbs/rc_verbs_iface.c b/src/uct/ib/rc/verbs/rc_verbs_iface.c index 8dccafa69eb..59b6d38953b 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_iface.c +++ b/src/uct/ib/rc/verbs/rc_verbs_iface.c @@ -50,6 +50,19 @@ static ucs_config_field_t uct_rc_verbs_iface_config_table[] = { "a minimum between this value and the TX queue length. -1 means no limit.", ucs_offsetof(uct_rc_verbs_iface_config_t, tx_max_wr), UCS_CONFIG_TYPE_UINT}, + {"RX_MAX_WR", "128", + "Number of receive WRs posted to each QP created without an SRQ, in the\n" + "range 1 to 65535, and limited by the device's max_qp_wr. Receive-buffer\n" + "and RX CQ requirements grow with the number of endpoints.", + ucs_offsetof(uct_rc_verbs_iface_config_t, rx_max_wr), UCS_CONFIG_TYPE_UINT}, + + {"SRQ_ENABLE", "try", + "Whether QPs use an SRQ:\n" + " - y|try : Use an SRQ if the device supports it, y warns if not\n" + " - n : Create QPs without an SRQ", + ucs_offsetof(uct_rc_verbs_iface_config_t, srq_enable), + UCS_CONFIG_TYPE_TERNARY}, + {"FLUSH_MODE", "auto", "Method to use for posting flush operation:\n" " - write0 : Post empty RDMA_WRITE\n" @@ -166,12 +179,13 @@ uct_rc_verbs_iface_poll_tx(uct_rc_verbs_iface_t *iface) return num_wcs; } -static unsigned uct_rc_verbs_iface_progress(void *arg) +static UCS_F_ALWAYS_INLINE unsigned +uct_rc_verbs_iface_progress(void *arg, int nosrq) { uct_rc_verbs_iface_t *iface = arg; unsigned count; - count = uct_rc_verbs_iface_poll_rx_common(iface); + count = uct_rc_verbs_iface_poll_rx(iface, nosrq); if (!uct_rc_iface_poll_tx(&iface->super, count)) { return count; } @@ -179,6 +193,16 @@ static unsigned uct_rc_verbs_iface_progress(void *arg) return count + uct_rc_verbs_iface_poll_tx(iface); } +static unsigned uct_rc_verbs_iface_progress_srq(void *arg) +{ + return uct_rc_verbs_iface_progress(arg, 0); +} + +static unsigned uct_rc_verbs_iface_progress_nosrq(void *arg) +{ + return uct_rc_verbs_iface_progress(arg, 1); +} + static void uct_rc_verbs_iface_init_inl_wrs(uct_rc_verbs_iface_t *iface) { memset(&iface->inl_am_wr, 0, sizeof(iface->inl_am_wr)); @@ -231,6 +255,11 @@ uct_rc_iface_verbs_init_rx(uct_rc_iface_t *rc_iface, { uct_rc_verbs_iface_t *iface = ucs_derived_of(rc_iface, uct_rc_verbs_iface_t); + if (rc_iface->config.srq_disable) { + iface->srq = NULL; + return UCS_OK; + } + return uct_rc_iface_init_rx(rc_iface, config, &iface->srq); } @@ -238,16 +267,40 @@ void uct_rc_iface_verbs_cleanup_rx(uct_rc_iface_t *rc_iface) { uct_rc_verbs_iface_t *iface = ucs_derived_of(rc_iface, uct_rc_verbs_iface_t); + if (rc_iface->config.srq_disable) { + return; + } + /* TODO flush RX buffers */ uct_ib_destroy_srq(iface->srq); } +/* Finalize cleanup after all receive WR flush completions have been polled. + * QPs attached to an SRQ are cleaned up by the LAST_WQE event instead. */ +void uct_rc_verbs_iface_gc_drain_check(uct_rc_verbs_iface_t *iface) +{ + uct_rc_verbs_iface_qp_cleanup_ctx_t *cleanup_ctx, *tmp; + + ucs_assert(iface->super.config.srq_disable); + + ucs_list_for_each_safe(cleanup_ctx, tmp, &iface->super.qp_gc_list, + super.list) { + if (cleanup_ctx->rx_remaining == 0) { + uct_rc_iface_qp_cleanup_progress(&cleanup_ctx->super); + } + } +} + static void uct_rc_verbs_iface_qp_cleanup(uct_rc_iface_qp_cleanup_ctx_t *rc_cleanup_ctx) { uct_rc_verbs_iface_qp_cleanup_ctx_t *cleanup_ctx = ucs_derived_of(rc_cleanup_ctx, uct_rc_verbs_iface_qp_cleanup_ctx_t); + uct_rc_verbs_iface_t *iface = ucs_derived_of(rc_cleanup_ctx->iface, + uct_rc_verbs_iface_t); + uct_ib_destroy_qp(cleanup_ctx->qp); + iface->rx_cq_available += iface->config.rx_max_wr; } static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, @@ -260,6 +313,8 @@ static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, uct_ib_md_t *ib_md = ucs_derived_of(tl_md, uct_ib_md_t); uct_rc_iface_init_attr_t init_attr = {}; uct_ib_qp_attr_t attr = {}; + unsigned rx_wr_limit, rx_batch_limit, num_eps; + unsigned rx_max_wr, rx_cq_len; const char *dev_name; ucs_status_t status; struct ibv_qp *qp; @@ -269,10 +324,46 @@ static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, return UCS_ERR_UNSUPPORTED; } + /* Use an SRQ unless disabled or unsupported by the device. */ + init_attr.srq_disable = (config->srq_enable == UCS_NO) || + !uct_ib_device_has_srq(&ib_md->dev); + if ((config->srq_enable == UCS_YES) && init_attr.srq_disable) { + ucs_warn("%s: SRQ is not supported, using QPs without an SRQ instead.", + uct_ib_device_name(&ib_md->dev)); + } + + if (init_attr.srq_disable) { + rx_wr_limit = ucs_min(IBV_DEV_ATTR(&ib_md->dev, max_qp_wr), UINT16_MAX); + if ((config->rx_max_wr == 0) || (config->rx_max_wr > rx_wr_limit)) { + ucs_error(UCS_DEFAULT_ENV_PREFIX "RC_VERBS_RX_MAX_WR must be " + "between 1 and %u when QPs are created without an SRQ", + rx_wr_limit); + return UCS_ERR_INVALID_PARAM; + } + + rx_max_wr = config->rx_max_wr; + rx_batch_limit = ucs_min(ib_config->rx.max_batch, + ucs_max(rx_max_wr / 4, 1)); + /* Room for the receive WRs of all endpoints; the 64-bit endpoint + * count is capped by the largest CQ before multiplying. */ + num_eps = ucs_min(ib_config->super.max_num_eps, + IBV_DEV_ATTR(&ib_md->dev, max_cqe) / rx_max_wr); + rx_cq_len = ucs_max(num_eps * rx_max_wr, ib_config->rx.queue_len); + ucs_debug("%s: RX CQ length %u for %u receive WRs per endpoint", + uct_ib_device_name(&ib_md->dev), rx_cq_len, rx_max_wr); + + /* Leave a batch of receive WRs unreposted, to avoid RNR. */ + init_attr.fc_max_wnd_size = ucs_max(rx_max_wr - rx_batch_limit, 1); + } else { + rx_max_wr = 0; + rx_batch_limit = 0; + rx_cq_len = ib_config->rx.queue_len; + } + init_attr.super.fc_req_size = sizeof(uct_rc_pending_req_t); init_attr.super.rx_hdr_len = sizeof(uct_rc_hdr_t); init_attr.super.qp_type = IBV_QPT_RC; - init_attr.super.cq_len[UCT_IB_DIR_RX] = ib_config->rx.queue_len; + init_attr.super.cq_len[UCT_IB_DIR_RX] = rx_cq_len; init_attr.super.cq_len[UCT_IB_DIR_TX] = config->super.tx_cq_len; init_attr.super.seg_size = ib_config->seg_size; init_attr.super.xport_hdr_len = ucs_max(sizeof(uct_rc_hdr_t), @@ -286,15 +377,26 @@ static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, &uct_rc_verbs_iface_ops, tl_md, worker, params, &config->super.super, &init_attr); + self->config.rx_max_wr = rx_max_wr; + self->rx_cq_available = rx_cq_len; + self->rx_post_recv_pending = 0; self->config.tx_max_wr = ucs_min(config->tx_max_wr, self->super.config.tx_qp_len); self->super.config.tx_moderation = ucs_min(self->super.config.tx_moderation, self->config.tx_max_wr / 4); self->super.config.fence_mode = (uct_rc_fence_mode_t)config->super.super.fence_mode; - self->super.progress = uct_rc_verbs_iface_progress; + self->super.progress = self->super.config.srq_disable ? + uct_rc_verbs_iface_progress_nosrq : + uct_rc_verbs_iface_progress_srq; self->super.super.config.sl = uct_ib_iface_config_select_sl(ib_config); uct_ib_iface_set_reverse_sl(&self->super.super, ib_config); + /* A QP must be able to reach the repost threshold. */ + if (self->super.config.srq_disable) { + self->super.super.config.rx_max_batch = + ucs_min(self->super.super.config.rx_max_batch, rx_batch_limit); + } + if ((config->super.super.fence_mode == UCT_RC_FENCE_MODE_WEAK) || (config->super.super.fence_mode == UCT_RC_FENCE_MODE_AUTO)) { self->super.config.fence_mode = UCT_RC_FENCE_MODE_WEAK; @@ -351,8 +453,8 @@ static UCS_CLASS_INIT_FUNC(uct_rc_verbs_iface_t, uct_md_h tl_md, /* Create a dummy QP in order to find out max_inline */ status = uct_rc_iface_qp_create(&self->super, &qp, &attr, - self->super.config.tx_qp_len, 0, - self->srq); + self->super.config.tx_qp_len, + self->config.rx_max_wr, self->srq); if (status != UCS_OK) { goto err_common_cleanup; } @@ -390,7 +492,7 @@ uct_rc_verbs_iface_common_prepost_recvs(uct_rc_verbs_iface_t *iface) iface->super.rx.srq.available = iface->super.rx.srq.quota; iface->super.rx.srq.quota = 0; while (iface->super.rx.srq.available > 0) { - if (uct_rc_verbs_iface_post_recv_common(iface, 1) == 0) { + if (uct_rc_verbs_iface_post_recv_common(iface, NULL, 1) == 0) { ucs_error("failed to post receives"); return UCS_ERR_NO_MEMORY; } @@ -416,7 +518,9 @@ void uct_rc_verbs_iface_common_progress_enable(uct_iface_h tl_iface, unsigned fl flags); } -unsigned uct_rc_verbs_iface_post_recv_always(uct_rc_verbs_iface_t *iface, unsigned max) +unsigned uct_rc_verbs_iface_post_recv_always(uct_rc_verbs_iface_t *iface, + uct_rc_verbs_ep_t *ep, + unsigned max) { struct ibv_recv_wr *bad_wr; uct_ib_recv_wr_t *wrs; @@ -431,15 +535,42 @@ unsigned uct_rc_verbs_iface_post_recv_always(uct_rc_verbs_iface_t *iface, unsign return 0; } - ret = ibv_post_srq_recv(iface->srq, &wrs[0].ibwr, &bad_wr); - if (ret != 0) { - ucs_fatal("ibv_post_srq_recv() returned %d: %m", ret); + if (ep == NULL) { + ret = ibv_post_srq_recv(iface->srq, &wrs[0].ibwr, &bad_wr); + if (ret != 0) { + ucs_fatal("ibv_post_srq_recv() returned %d: %m", ret); + } + iface->super.rx.srq.available -= count; + } else { + ret = ibv_post_recv(ep->qp, &wrs[0].ibwr, &bad_wr); + if (ret != 0) { + ucs_fatal("ibv_post_recv() on QP 0x%x returned %d: %m", + ep->qp->qp_num, ret); + } + ep->rx_available -= count; } - iface->super.rx.srq.available -= count; return count; } +void uct_rc_verbs_iface_eps_post_recv(uct_rc_verbs_iface_t *iface) +{ + unsigned batch = iface->super.super.config.rx_max_batch; + uct_rc_verbs_ep_t *ep; + uct_rc_ep_t *rc_ep; + + iface->rx_post_recv_pending = 0; + ucs_list_for_each(rc_ep, &iface->super.ep_list, list) { + ep = ucs_derived_of(rc_ep, uct_rc_verbs_ep_t); + if (!uct_rc_verbs_ep_can_post_recv(ep) || (ep->rx_available < batch)) { + continue; + } + + /* Requeue the scan if the pool cannot supply a full batch. */ + uct_rc_verbs_iface_post_recv_common(iface, ep, 0); + } +} + static UCS_CLASS_CLEANUP_FUNC(uct_rc_verbs_iface_t) { uct_base_iface_progress_disable(&self->super.super.super.super, @@ -592,8 +723,9 @@ uct_rc_verbs_query_tl_devices(uct_md_h md, return status; } - return uct_ib_device_query_ports(&ib_md->dev, UCT_IB_DEVICE_FLAG_SRQ, - tl_devices_p, num_tl_devices_p); + /* Devices without SRQ support can use receive WRs posted to each QP. */ + return uct_ib_device_query_ports(&ib_md->dev, 0, tl_devices_p, + num_tl_devices_p); } UCT_TL_DEFINE_ENTRY(&uct_ib_component, rc_verbs, uct_rc_verbs_query_tl_devices, diff --git a/src/uct/ib/rc/verbs/rc_verbs_impl.h b/src/uct/ib/rc/verbs/rc_verbs_impl.h index acd743dd8f1..94a6d94a78a 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_impl.h +++ b/src/uct/ib/rc/verbs/rc_verbs_impl.h @@ -26,24 +26,43 @@ uct_rc_verbs_iface_common_prepost_recvs(uct_rc_verbs_iface_t *iface); void uct_rc_verbs_iface_common_progress_enable(uct_iface_h tl_iface, unsigned flags); -unsigned uct_rc_verbs_iface_post_recv_always(uct_rc_verbs_iface_t *iface, unsigned max); +void uct_rc_verbs_iface_gc_drain_check(uct_rc_verbs_iface_t *iface); -static inline unsigned uct_rc_verbs_iface_post_recv_common(uct_rc_verbs_iface_t *iface, - int fill) +void uct_rc_verbs_iface_eps_post_recv(uct_rc_verbs_iface_t *iface); + +unsigned uct_rc_verbs_iface_post_recv_always(uct_rc_verbs_iface_t *iface, + uct_rc_verbs_ep_t *ep, + unsigned max); + +static inline unsigned +uct_rc_verbs_iface_post_recv_common(uct_rc_verbs_iface_t *iface, + uct_rc_verbs_ep_t *ep, int fill) { - unsigned batch = iface->super.super.config.rx_max_batch; - unsigned count; + unsigned available = (ep == NULL) ? iface->super.rx.srq.available : + ep->rx_available; + unsigned batch = iface->super.super.config.rx_max_batch; + unsigned count, posted; - if (iface->super.rx.srq.available < batch) { + if (available < batch) { if (ucs_likely(fill == 0)) { return 0; } else { - count = iface->super.rx.srq.available; + count = available; } } else { count = batch; } - return uct_rc_verbs_iface_post_recv_always(iface, count); + + posted = uct_rc_verbs_iface_post_recv_always(iface, ep, count); + + /* A QP with no posted receive WRs cannot generate a receive completion. + * Retry from progress when the shared receive buffer pool cannot supply a + * full batch. */ + if ((ep != NULL) && (posted < count) && (ep->rx_available >= batch)) { + iface->rx_post_recv_pending = 1; + } + + return posted; } static UCS_F_ALWAYS_INLINE void @@ -74,8 +93,54 @@ uct_rc_verbs_iface_handle_am(uct_rc_iface_t *iface, uct_rc_hdr_t *hdr, } } +/* Account for a receive WR flush completion after its QP was removed from the + * lookup table. A QP created without an SRQ is kept until every posted receive + * WR has generated a completion. */ +static UCS_F_ALWAYS_INLINE void +uct_rc_verbs_iface_gc_drain_cqe(uct_rc_verbs_iface_t *iface, uint32_t qp_num) +{ + uct_rc_verbs_iface_qp_cleanup_ctx_t *cleanup_ctx; + + ucs_list_for_each(cleanup_ctx, &iface->super.qp_gc_list, super.list) { + if (cleanup_ctx->super.qp_num == qp_num) { + ucs_assertv(cleanup_ctx->rx_remaining > 0, "qp_num=0x%x", qp_num); + cleanup_ctx->rx_remaining--; + return; + } + } +} + +/* An endpoint which was cancelled or failed must not receive any more */ +static UCS_F_ALWAYS_INLINE int +uct_rc_verbs_ep_can_post_recv(const uct_rc_verbs_ep_t *ep) +{ + return !(ep->super.flags & (UCT_RC_EP_FLAG_FLUSH_CANCEL | + UCT_RC_EP_FLAG_ERR_HANDLER_INVOKED)); +} + +/* Account for a receive completion of a QP without an SRQ. */ +static UCS_F_ALWAYS_INLINE void +uct_rc_verbs_iface_rx_complete_nosrq(uct_rc_verbs_iface_t *iface, + uint32_t qp_num, int repost) +{ + uct_rc_ep_t *rc_ep = uct_rc_iface_lookup_ep(&iface->super, qp_num); + uct_rc_verbs_ep_t *ep = ucs_derived_of(rc_ep, uct_rc_verbs_ep_t); + + if (ep == NULL) { + uct_rc_verbs_iface_gc_drain_cqe(iface, qp_num); + return; + } + + ep->rx_available++; + if (repost && uct_rc_verbs_ep_can_post_recv(ep)) { + uct_rc_verbs_iface_post_recv_common(iface, ep, 0); + } +} + +/* Poll receive completions. When nosrq is set, receive WRs are posted to every + * QP separately, so the QP number of a completion selects the receive queue. */ static UCS_F_ALWAYS_INLINE unsigned -uct_rc_verbs_iface_poll_rx_common(uct_rc_verbs_iface_t *iface) +uct_rc_verbs_iface_poll_rx(uct_rc_verbs_iface_t *iface, int nosrq) { uct_ib_iface_recv_desc_t *desc; uct_rc_hdr_t *hdr; @@ -95,11 +160,17 @@ uct_rc_verbs_iface_poll_rx_common(uct_rc_verbs_iface_t *iface) hdr = (uct_rc_hdr_t *)uct_ib_iface_recv_desc_hdr(&iface->super.super, desc); if (ucs_unlikely(wc[i].status != IBV_WC_SUCCESS)) { /* A failed receive does not deliver an active message, so return - * its descriptor to the pool. A flushed one arrives during - * endpoint destruction. */ + * its descriptor to the pool. A flushed WR belongs to a QP which is + * being destroyed, so post nothing to it, while an aborted one + * leaves the QP usable. */ if ((wc[i].status == IBV_WC_REM_ABORT_ERR) || (wc[i].status == IBV_WC_WR_FLUSH_ERR)) { ucs_mpool_put_inline(desc); + if (nosrq) { + uct_rc_verbs_iface_rx_complete_nosrq(iface, wc[i].qp_num, + wc[i].status != + IBV_WC_WR_FLUSH_ERR); + } continue; } UCT_IB_IFACE_VERBS_COMPLETION_FATAL("receive", &iface->super.super, i, wc); @@ -110,13 +181,30 @@ uct_rc_verbs_iface_poll_rx_common(uct_rc_verbs_iface_t *iface) uct_rc_ep_packet_dump); uct_rc_verbs_iface_handle_am(&iface->super, hdr, wc[i].wr_id, wc[i].qp_num, wc[i].byte_len, wc[i].imm_data, wc[i].slid); + + if (nosrq) { + uct_rc_verbs_iface_rx_complete_nosrq(iface, wc[i].qp_num, 1); + } + } + if (!nosrq) { + iface->super.rx.srq.available += num_wcs; } - iface->super.rx.srq.available += num_wcs; UCS_STATS_UPDATE_COUNTER(iface->super.super.stats, UCT_IB_IFACE_STAT_RX_COMPLETION, num_wcs); out: - uct_rc_verbs_iface_post_recv_common(iface, 0); + if (nosrq) { + if (ucs_unlikely(iface->rx_post_recv_pending)) { + uct_rc_verbs_iface_eps_post_recv(iface); + } + + if (ucs_unlikely(!ucs_list_is_empty(&iface->super.qp_gc_list))) { + uct_rc_verbs_iface_gc_drain_check(iface); + } + } else { + uct_rc_verbs_iface_post_recv_common(iface, NULL, 0); + } + return num_wcs; } diff --git a/test/apps/test_fuzzy_match.py b/test/apps/test_fuzzy_match.py index 41d609f1ee9..0f16a8b9a43 100755 --- a/test/apps/test_fuzzy_match.py +++ b/test/apps/test_fuzzy_match.py @@ -102,7 +102,7 @@ def has_ib(): {'UCX_RLS' : ['UCX_TLS']}] if has_ib(): - test_cases += [{'UCX_RC_VERBS_RX_MAX_BUF' : ['UCX_RC_VERBS_TX_MAX_BUFS', 'UCX_RC_VERBS_RX_MAX_BUFS', 'UCX_UD_VERBS_RX_MAX_BUFS']}] + test_cases += [{'UCX_RC_VERBS_RX_MAX_BUF' : ['UCX_RC_VERBS_TX_MAX_BUFS', 'UCX_RC_VERBS_RX_MAX_BUFS', 'UCX_RC_VERBS_RX_MAX_WR', 'UCX_UD_VERBS_RX_MAX_BUFS']}] for test_case in test_cases: runner.run(test_case) From ac194c17379d56ba8726824d84b79aef1ac8790f Mon Sep 17 00:00:00 2001 From: Guangguan Wang Date: Wed, 9 Sep 2026 20:27:20 +0800 Subject: [PATCH 8/8] GTEST/UCT/IB: Skip LAST_WQE tests when the QPs have no SRQ The LAST_WQE_REACHED event is generated only for a QP which is attached to an SRQ, so the tests which wait for it never complete otherwise. Skip them when the interface works without an SRQ. Signed-off-by: Guangguan Wang --- test/gtest/uct/ib/test_ib_event.cc | 39 ++++++++++++++++++++++++++---- 1 file changed, 34 insertions(+), 5 deletions(-) diff --git a/test/gtest/uct/ib/test_ib_event.cc b/test/gtest/uct/ib/test_ib_event.cc index 049629268cb..20686557971 100644 --- a/test/gtest/uct/ib/test_ib_event.cc +++ b/test/gtest/uct/ib/test_ib_event.cc @@ -25,6 +25,25 @@ class uct_test_event_base : public uct_p2p_test { public: uct_test_event_base(): uct_p2p_test(0), m_event() {} + /* LAST_WQE_REACHED is generated only for QPs attached to an SRQ. */ + static bool iface_has_srq(entity &e) { + uct_rc_iface_t *iface = ucs_derived_of(e.iface(), uct_rc_iface_t); + + return !iface->config.srq_disable; + } + + /* Before init(), no entity exists and the test cannot determine whether + * its QPs use an SRQ, so do not skip yet. */ + bool has_srq() { + FOR_EACH_ENTITY(iter) { + if (!iface_has_srq(**iter)) { + return false; + } + } + + return true; + } + class qp { public: qp(entity &e) : m_e(e) {} @@ -217,7 +236,7 @@ class uct_p2p_test_event_log : public uct_ep_test_event { } }; -UCS_TEST_P(uct_p2p_test_event_log, last_wqe) +UCS_TEST_SKIP_COND_P(uct_p2p_test_event_log, last_wqe, !has_srq()) { const p2p_resource *r = dynamic_cast(GetParam()); ucs_assert_always(r != NULL); @@ -255,7 +274,8 @@ class uct_p2p_test_event : public uct_ep_test_event { }; -UCS_TEST_P(uct_p2p_test_event, last_wqe_cb_after_subscribe) +UCS_TEST_SKIP_COND_P(uct_p2p_test_event, last_wqe_cb_after_subscribe, + !has_srq()) { const p2p_resource *r = dynamic_cast(GetParam()); ucs_assert_always(r != NULL); @@ -266,7 +286,8 @@ UCS_TEST_P(uct_p2p_test_event, last_wqe_cb_after_subscribe) } } -UCS_TEST_P(uct_p2p_test_event, last_wqe_cb_before_subscribe) +UCS_TEST_SKIP_COND_P(uct_p2p_test_event, last_wqe_cb_before_subscribe, + !has_srq()) { const p2p_resource *r = dynamic_cast(GetParam()); ucs_assert_always(r != NULL); @@ -304,6 +325,12 @@ class uct_qp_test_event : public uct_test_event_base { virtual void init() { uct_test::init(); m_e.reset(create_entity(0)); + check_skip_test(); + } + + /* m_e is not stored in m_entities, so check its interface directly. */ + bool has_srq() { + return (m_e == NULL) || iface_has_srq(*m_e); } bool wait_for_last_wqe_event(bool before) { @@ -418,12 +445,14 @@ class uct_qp_test_event : public uct_test_event_base { }; }; -UCS_TEST_P(uct_qp_test_event, last_wqe_cb_after_subscribe) +UCS_TEST_SKIP_COND_P(uct_qp_test_event, last_wqe_cb_after_subscribe, + !has_srq()) { ASSERT_TRUE(wait_for_last_wqe_event(false)); } -UCS_TEST_P(uct_qp_test_event, last_wqe_cb_before_subscribe) +UCS_TEST_SKIP_COND_P(uct_qp_test_event, last_wqe_cb_before_subscribe, + !has_srq()) { ASSERT_TRUE(wait_for_last_wqe_event(true)); }