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 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); 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..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); @@ -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..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); } @@ -773,19 +772,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.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/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_ep.c b/src/uct/ib/rc/base/rc_ep.c index 92da16369e5..2034498adc5 100644 --- a/src/uct/ib/rc/base/rc_ep.c +++ b/src/uct/ib/rc/base/rc_ep.c @@ -118,20 +118,29 @@ 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; - 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, @@ -186,6 +195,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); diff --git a/src/uct/ib/rc/base/rc_iface.c b/src/uct/ib/rc/base/rc_iface.c index d2deba0cd47..01924bdba85 100644 --- a/src/uct/ib/rc/base/rc_iface.c +++ b/src/uct/ib/rc/base/rc_iface.c @@ -572,28 +572,32 @@ 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; 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; 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); 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; - 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 +624,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 +707,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; @@ -720,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 { @@ -753,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); @@ -810,11 +819,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; @@ -826,9 +836,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 737fc4de75d..fbc28441023 100644 --- a/src/uct/ib/rc/base/rc_iface.h +++ b/src/uct/ib/rc/base/rc_iface.h @@ -250,6 +250,16 @@ typedef struct uct_rc_srq { } uct_rc_srq_t; +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; + + struct uct_rc_iface { uct_ib_iface_t super; @@ -299,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; @@ -328,7 +340,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 { @@ -408,11 +420,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.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 77d35212295..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,8 +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, iface->srq); + iface->super.config.tx_qp_len, + iface->config.rx_max_wr, iface->srq); if (status != UCS_OK) { goto err; } @@ -709,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); @@ -721,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); @@ -736,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: @@ -760,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 f548cdd3968..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, @@ -258,8 +311,10 @@ 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 = {}; + 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,30 +324,79 @@ 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; + /* 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] = 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), + 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, &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; @@ -350,7 +454,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->srq); + self->config.rx_max_wr, self->srq); if (status != UCS_OK) { goto err_common_cleanup; } @@ -388,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; } @@ -414,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; @@ -429,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, @@ -590,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 14f0d24e9bd..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; @@ -94,11 +159,18 @@ 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 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); @@ -109,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) diff --git a/test/gtest/uct/ib/test_ib_event.cc b/test/gtest/uct/ib/test_ib_event.cc index 91bab78af67..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) { @@ -387,7 +414,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); @@ -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)); }