diff --git a/src/uct/ib/mlx5/ib_mlx5.c b/src/uct/ib/mlx5/ib_mlx5.c index e9d9255fad8..711eb47cb6e 100644 --- a/src/uct/ib/mlx5/ib_mlx5.c +++ b/src/uct/ib/mlx5/ib_mlx5.c @@ -858,6 +858,23 @@ void *uct_ib_mlx5_txwq_get_wqe(const uct_ib_mlx5_txwq_t *txwq, uint16_t pi) return UCS_PTR_BYTE_OFFSET(txwq->qstart, (pi % num_bb) * MLX5_SEND_WQE_BB); } +size_t uct_ib_mlx5_wqe_size(const struct mlx5_wqe_ctrl_seg *ctrl) +{ + uint8_t ds = ntohl(ctrl->qpn_ds) & UINT8_MAX; + + ucs_assertv_always( + (ds > 0) && + ((ds * UCT_IB_MLX5_WQE_SEG_SIZE) <= + UCT_IB_MLX5_MAX_SEND_WQE_SIZE), + "ds=%u", ds); + return ds * UCT_IB_MLX5_WQE_SEG_SIZE; +} + +uint16_t uct_ib_mlx5_txwq_next_wqe_index(uint16_t index, size_t wqe_size) +{ + return index + ucs_div_round_up(wqe_size, MLX5_SEND_WQE_BB); +} + uint16_t uct_ib_mlx5_txwq_num_posted_wqes(const uct_ib_mlx5_txwq_t *txwq, uint16_t outstanding) { @@ -873,8 +890,8 @@ uint16_t uct_ib_mlx5_txwq_num_posted_wqes(const uct_ib_mlx5_txwq_t *txwq, ucs_assert(pi == txwq->hw_ci); do { ctrl = uct_ib_mlx5_txwq_get_wqe(txwq, pi); - wqe_size = (ctrl->qpn_ds >> 24) * UCT_IB_MLX5_WQE_SEG_SIZE; - pi += (wqe_size + MLX5_SEND_WQE_BB - 1) / MLX5_SEND_WQE_BB; + wqe_size = uct_ib_mlx5_wqe_size(ctrl); + pi = uct_ib_mlx5_txwq_next_wqe_index(pi, wqe_size); ++count; } while (pi != txwq->sw_pi); diff --git a/src/uct/ib/mlx5/ib_mlx5.h b/src/uct/ib/mlx5/ib_mlx5.h index 3732e228aa9..c261dbcd724 100644 --- a/src/uct/ib/mlx5/ib_mlx5.h +++ b/src/uct/ib/mlx5/ib_mlx5.h @@ -64,6 +64,8 @@ #define UCT_IB_MLX5_CQ_SET_CI 0 #define UCT_IB_MLX5_CQ_ARM_DB 1 #define UCT_IB_MLX5_LOG_MAX_MSG_SIZE 30 +#define UCT_IB_MLX5_PSN_BITS 24 +#define UCT_IB_MLX5_PSN_MASK UCS_MASK(UCT_IB_MLX5_PSN_BITS) #define UCT_IB_MLX5_ATOMIC_MODE_COMP 1 #define UCT_IB_MLX5_ATOMIC_MODE_EXT 3 #define UCT_IB_MLX5_CQE_FLAG_L3_IN_DATA UCS_BIT(28) /* GRH/IP in the receive buffer */ @@ -709,7 +711,7 @@ typedef struct uct_ib_mlx5_txwq { static UCS_F_ALWAYS_INLINE uint32_t uct_ib_mlx5_txwq_get_next_wqe_psn(const uct_ib_mlx5_txwq_t *txwq) { - return txwq->next_wqe_psn & UCS_MASK(24); + return txwq->next_wqe_psn & UCT_IB_MLX5_PSN_MASK; } @@ -917,6 +919,12 @@ ucs_status_t uct_ib_mlx5_txwq_init(uct_priv_worker_t *worker, /* Get pointer to a WQE by producer index */ void *uct_ib_mlx5_txwq_get_wqe(const uct_ib_mlx5_txwq_t *txwq, uint16_t pi); +/* Get the WQE size in bytes from its control segment */ +size_t uct_ib_mlx5_wqe_size(const struct mlx5_wqe_ctrl_seg *ctrl); + +/* Get the index of the WQE that follows a WQE of the given size */ +uint16_t uct_ib_mlx5_txwq_next_wqe_index(uint16_t index, size_t wqe_size); + /* Count how many WQEs are currently posted */ uint16_t uct_ib_mlx5_txwq_num_posted_wqes(const uct_ib_mlx5_txwq_t *txwq, uint16_t outstanding); diff --git a/src/uct/ib/mlx5/ib_mlx5_ext.c b/src/uct/ib/mlx5/ib_mlx5_ext.c index 643e0dbc21c..ede800d8bb5 100644 --- a/src/uct/ib/mlx5/ib_mlx5_ext.c +++ b/src/uct/ib/mlx5/ib_mlx5_ext.c @@ -111,55 +111,6 @@ uct_ib_mlx5_ext_ep_put_sgl_zcopy(uct_ep_h ep, void * const *buffers, return UCS_ERR_UNSUPPORTED; } -static ucs_status_t uct_ib_mlx5_ext_ep_outstanding_purge_check_param( - const uct_ep_outstanding_purge_params_t *params) -{ - if (params == NULL) { - ucs_error("ib mlx5 ext: outstanding purge parameters are NULL"); - return UCS_ERR_INVALID_PARAM; - } - - if (!(params->field_mask & UCT_EP_OUTSTANDING_FIELD_RX_TOKEN) || - !(params->field_mask & UCT_EP_OUTSTANDING_FIELD_CB) || - (params->rx_token == NULL) || (params->cb == NULL)) { - ucs_error("ib mlx5 ext: rx token or callback is not set or is NULL"); - return UCS_ERR_INVALID_PARAM; - } - - return UCS_OK; -} - -ucs_status_t uct_ib_mlx5_ext_ep_outstanding_purge( - uct_ep_h ep, const uct_ep_outstanding_purge_params_t *params) -{ - ucs_status_t status; - uct_ib_mlx5_ext_plugin_t *plugin; - - status = uct_ib_mlx5_ext_ep_outstanding_purge_check_param(params); - if (status != UCS_OK) { - return status; - } - - ucs_list_for_each(plugin, &uct_ib_mlx5_ext_plugins, list) { - if (ucs_unlikely(uct_ib_mlx5_ext_is_unsupported_op( - (const void*)plugin->ops.ep_outstanding_purge))) { - continue; - } - - /* The provider validates parameters and token signatures itself, and - * returns an error if it should not handle this request. */ - status = plugin->ops.ep_outstanding_purge(ep, params); - if (status != UCS_OK) { - continue; - } - - return UCS_OK; - } - - ucs_error("ib mlx5 ext: ep outstanding purge is not supported"); - return UCS_ERR_UNSUPPORTED; -} - void uct_ib_mlx5_ext_cleanup(void) { uct_ib_mlx5_ext_plugin_t *plugin, *tmp; @@ -211,7 +162,7 @@ ucs_status_t uct_ib_mlx5_ext_register(const uct_ib_mlx5_ext_ops_t *ops) num_plugins = ucs_list_length(&uct_ib_mlx5_ext_plugins); ucs_debug("ib mlx5 ext: registered plugin name=%s iface_query=%s " - "put_sgl_zcopy=%s outstanding_purge=%s (total=%u)", + "put_sgl_zcopy=%s (total=%u)", plugin->ops.name, uct_ib_mlx5_ext_is_unsupported_op( (const void*)plugin->ops.iface_query) ? @@ -221,10 +172,6 @@ ucs_status_t uct_ib_mlx5_ext_register(const uct_ib_mlx5_ext_ops_t *ops) (const void*)plugin->ops.ep_put_sgl_zcopy) ? "unsupported" : "supported", - uct_ib_mlx5_ext_is_unsupported_op( - (const void*)plugin->ops.ep_outstanding_purge) ? - "unsupported" : - "supported", num_plugins); return UCS_OK; } diff --git a/src/uct/ib/mlx5/ib_mlx5_ext.h b/src/uct/ib/mlx5/ib_mlx5_ext.h index 04a9d2a44b1..af8ad012b3e 100644 --- a/src/uct/ib/mlx5/ib_mlx5_ext.h +++ b/src/uct/ib/mlx5/ib_mlx5_ext.h @@ -77,7 +77,6 @@ typedef struct uct_ib_mlx5_ext_ops { uct_ib_mlx5_ext_iface_query_func_t iface_query; /**< Iface query callback */ uct_ib_mlx5_ext_max_put_sgl_zcopy_count_func_t max_put_sgl_zcopy_count; /**< Maximum PUT SGL zero-copy entry count callback */ uct_ep_put_sgl_zcopy_func_t ep_put_sgl_zcopy; /**< PUT SGL zero-copy callback */ - uct_ep_outstanding_purge_func_t ep_outstanding_purge; /**< Outstanding operation purge callback */ } uct_ib_mlx5_ext_ops_t; /** @@ -118,9 +117,6 @@ ucs_status_t uct_ib_mlx5_ext_ep_put_sgl_zcopy(uct_ep_h ep, size_t count, uct_completion_t *comp); -ucs_status_t uct_ib_mlx5_ext_ep_outstanding_purge( - uct_ep_h ep, const uct_ep_outstanding_purge_params_t *params); - END_C_DECLS #endif /* UCT_IB_MLX5_EXT_H_ */ diff --git a/src/uct/ib/mlx5/ib_mlx5_log.c b/src/uct/ib/mlx5/ib_mlx5_log.c index 8e1bc9fbfda..9277e1a6986 100644 --- a/src/uct/ib/mlx5/ib_mlx5_log.c +++ b/src/uct/ib/mlx5/ib_mlx5_log.c @@ -16,11 +16,6 @@ #include -static void uct_ib_mlx5_wqe_dump(uct_ib_iface_t *iface, void *wqe, void *qstart, - void *qend, int max_sge, int dump_qp, - uct_log_data_dump_func_t packet_dump_cb, - char *buffer, size_t max, uct_ib_log_sge_t *log_sge); - static void uct_ib_mlx5_resp_error_dump(const uct_ib_mlx5_srq_seg_t *seg, unsigned max_strides, char *buffer, size_t max); @@ -281,10 +276,10 @@ static size_t uct_ib_mlx5_dump_dgram(char *buf, size_t max, void *seg, int is_et UCT_IB_MLX5_AV_FULL_SIZE : UCT_IB_MLX5_AV_BASE_SIZE; } -static void uct_ib_mlx5_wqe_dump(uct_ib_iface_t *iface, void *wqe, void *qstart, - void *qend, int max_sge, int dump_qp, - uct_log_data_dump_func_t packet_dump_cb, - char *buffer, size_t max, uct_ib_log_sge_t *log_sge) +void uct_ib_mlx5_wqe_dump(uct_ib_iface_t *iface, void *wqe, void *qstart, + void *qend, int max_sge, int dump_qp, + uct_log_data_dump_func_t packet_dump_cb, + char *buffer, size_t max, uct_ib_log_sge_t *log_sge) { static uct_ib_opcode_t opcodes[] = { [MLX5_OPCODE_NOP] = { "NOP", 0 }, @@ -308,18 +303,26 @@ static void uct_ib_mlx5_wqe_dump(uct_ib_iface_t *iface, void *wqe, void *qstart, uint8_t opcode = ctrl->opmod_idx_opcode >> 24; uint8_t opmod = ctrl->opmod_idx_opcode & 0xff; uint32_t qp_num = ntohl(ctrl->qpn_ds) >> 8; - int ds = ctrl->qpn_ds >> 24; - uct_ib_opcode_t *op = &opcodes[opcode]; + int ds = ntohl(ctrl->qpn_ds) & UINT8_MAX; char *s = buffer; char *ends = buffer + max; - const char* sg_prefix_arr = (op->flags & UCT_IB_OPCODE_FLAG_HAS_DMA) ? - "GS" : NULL; + uct_ib_opcode_t *op; + const char *sg_prefix_arr; struct ibv_sge sg_list[16]; uint64_t inline_bitmap; int i, is_inline, is_eth; size_t dg_size; void *seg; + if ((opcode >= ucs_static_array_size(opcodes)) || + (opcodes[opcode].name == NULL)) { + snprintf(buffer, max, "unknown opcode 0x%x", opcode); + return; + } + + op = &opcodes[opcode]; + sg_prefix_arr = (op->flags & UCT_IB_OPCODE_FLAG_HAS_DMA) ? "GS" : NULL; + /* QP and WQE index */ if (dump_qp) { snprintf(s, ends - s, "QP 0x%x [%03ld] ", qp_num, @@ -387,8 +390,8 @@ static void uct_ib_mlx5_wqe_dump(uct_ib_iface_t *iface, void *wqe, void *qstart, /* Extended atomic segment */ if (op->flags & UCT_IB_OPCODE_FLAG_HAS_EXT_ATOMIC) { - uint64_t add, boundary, compare, swap, compare_mask, swap_mask; int size = 1 << ((opmod & 7) + 2); + uint64_t add, boundary, compare, swap, compare_mask, swap_mask; if (opcode == MLX5_OPCODE_ATOMIC_MASKED_FA) { add = network_to_host(seg, size); diff --git a/src/uct/ib/mlx5/ib_mlx5_log.h b/src/uct/ib/mlx5/ib_mlx5_log.h index 882dab56b96..233c940f843 100644 --- a/src/uct/ib/mlx5/ib_mlx5_log.h +++ b/src/uct/ib/mlx5/ib_mlx5_log.h @@ -24,6 +24,12 @@ ucs_status_t uct_ib_mlx5_completion_with_err(uct_ib_iface_t *iface, ucs_log_level_t log_level); +void uct_ib_mlx5_wqe_dump(uct_ib_iface_t *iface, void *wqe, void *qstart, + void *qend, int max_sge, int dump_qp, + uct_log_data_dump_func_t packet_dump_cb, + char *buffer, size_t max, uct_ib_log_sge_t *log_sge); + + void __uct_ib_mlx5_log_tx(const char *file, int line, const char *function, uct_ib_iface_t *iface, void *wqe, void *qstart, void *qend, int max_sge, uct_ib_log_sge_t *log_sge, diff --git a/src/uct/ib/mlx5/rc/rc_mlx5.h b/src/uct/ib/mlx5/rc/rc_mlx5.h index c20045fd2d1..03bb325eb16 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5.h +++ b/src/uct/ib/mlx5/rc/rc_mlx5.h @@ -209,6 +209,9 @@ ucs_status_t uct_rc_mlx5_base_ep_invalidate(uct_ep_h tl_ep, const uct_ep_invalidate_params_t *params); +ucs_status_t uct_rc_mlx5_ep_outstanding_purge( + uct_ep_h tl_ep, const uct_ep_outstanding_purge_params_t *params); + ucs_status_t uct_rc_mlx5_base_ep_fc_ctrl(uct_ep_t *tl_ep, unsigned op, uct_rc_pending_req_t *req); diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_ep.c b/src/uct/ib/mlx5/rc/rc_mlx5_ep.c index 6b809f8fc6d..56aac5aaec7 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_ep.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_ep.c @@ -817,6 +817,310 @@ ucs_status_t uct_rc_mlx5_base_ep_invalidate(uct_ep_h tl_ep, IBV_QPS_ERR); } +static uint8_t +uct_rc_mlx5_wqe_opcode(const struct mlx5_wqe_ctrl_seg *ctrl) +{ + return ctrl->opmod_idx_opcode >> 24; +} + +static const struct mlx5_wqe_inl_data_seg * +uct_rc_mlx5_wqe_get_inline_seg( + const uct_ib_mlx5_txwq_t *txwq, + const struct mlx5_wqe_ctrl_seg *ctrl, size_t wqe_size, + size_t *inline_length_p) +{ + const struct mlx5_wqe_inl_data_seg *inl; + uint32_t byte_count; + + ucs_assertv_always(wqe_size >= (sizeof(*ctrl) + sizeof(*inl)), + "wqe_size=%zu", wqe_size); + + inl = uct_ib_mlx5_txwq_wrap_any((uct_ib_mlx5_txwq_t*)txwq, + (void*)(ctrl + 1)); + byte_count = ntohl(inl->byte_count); + if (!(byte_count & MLX5_INLINE_SEG)) { + return NULL; + } + + *inline_length_p = byte_count & ~MLX5_INLINE_SEG; + return inl; +} + +static void +uct_rc_mlx5_txwq_copy_segs(const uct_ib_mlx5_txwq_t *txwq, const void *src, + void *dst, size_t length) +{ + size_t copy_len = ucs_min(length, UCS_PTR_BYTE_DIFF(src, txwq->qend)); + + memcpy(dst, src, copy_len); + if (copy_len < length) { + memcpy(UCS_PTR_BYTE_OFFSET(dst, copy_len), txwq->qstart, + length - copy_len); + } +} + +static ucs_status_t +uct_rc_mlx5_op_info_fill_am_short(const uct_ib_mlx5_txwq_t *txwq, + const struct mlx5_wqe_inl_data_seg *inl, + size_t inline_length, void *callback_data, + uct_ep_op_info_t *info) +{ + uct_rc_mlx5_am_short_hdr_t *am; + + uct_rc_mlx5_txwq_copy_segs(txwq, inl + 1, callback_data, inline_length); + + am = callback_data; + if ((am->rc_hdr.rc_hdr.am_id & UCT_RC_EP_FC_MASK) == + UCT_RC_EP_FC_PURE_GRANT) { + return UCS_ERR_NO_ELEM; + } + + if (inline_length < sizeof(*am)) { + return UCS_ERR_UNSUPPORTED; + } + + info->field_mask = UCT_EP_OP_INFO_FIELD_OPERATION | + UCT_EP_OP_INFO_FIELD_AM; + info->operation = UCT_EP_OP_AM_SHORT; + info->am.field_mask = UCT_EP_OP_INFO_AM_FIELD_AM_ID | + UCT_EP_OP_INFO_AM_FIELD_HEADER_VALUE | + UCT_EP_OP_INFO_AM_FIELD_PAYLOAD_DATA; + info->am.am_id = am->rc_hdr.rc_hdr.am_id & ~UCT_RC_EP_FC_MASK; + info->am.header.value = am->am_hdr; + info->am.payload.data.buffer = am + 1; + info->am.payload.data.length = inline_length - sizeof(*am); + return UCS_OK; +} + +static ucs_status_t uct_rc_mlx5_op_info_fill_am( + const uct_ib_mlx5_txwq_t *txwq, + const struct mlx5_wqe_ctrl_seg *ctrl, size_t wqe_size, + void *callback_data, uct_ep_op_info_t *info) +{ + const struct mlx5_wqe_inl_data_seg *inl; + size_t inline_length; + + memset(info, 0, sizeof(*info)); + ucs_assert(uct_rc_mlx5_wqe_opcode(ctrl) == MLX5_OPCODE_SEND); + + inl = uct_rc_mlx5_wqe_get_inline_seg(txwq, ctrl, wqe_size, + &inline_length); + if (inl != NULL) { + return uct_rc_mlx5_op_info_fill_am_short( + txwq, inl, inline_length, callback_data, info); + } + + return UCS_ERR_UNSUPPORTED; +} + +static int uct_ib_mlx5_wqe_is_delivered(uint32_t wqe_first_psn, + uint32_t receiver_next_psn, + uint32_t num_packets) +{ + const uint32_t psn_half = UCS_BIT(UCT_IB_MLX5_PSN_BITS - 1); + uint32_t diff; + + ucs_assert(num_packets > 0); + + /* + * PSNs wrap in a 24-bit sequence space. Since the outstanding window is + * smaller than half of that space, the masked forward distance determines + * their order: 0 < diff < psn_half means receiver_next_psn is ahead of + * wqe_first_psn, while diff > psn_half means it is behind and the + * subtraction wrapped. The validated outstanding window excludes the + * ambiguous distance of exactly psn_half. + */ + diff = (receiver_next_psn - wqe_first_psn) & UCT_IB_MLX5_PSN_MASK; + return (diff < psn_half) && (diff >= num_packets); +} + +static UCS_F_NOINLINE UCS_F_NORETURN void +uct_rc_mlx5_wqe_unsupported(uct_ib_iface_t *iface, + const uct_ib_mlx5_txwq_t *txwq, + const struct mlx5_wqe_ctrl_seg *ctrl, + size_t wqe_size) +{ + char wqe_dump[256] = {0}; + + uct_ib_mlx5_wqe_dump(iface, (void*)ctrl, txwq->qstart, txwq->qend, + INT_MAX, 0, NULL, wqe_dump, sizeof(wqe_dump) - 1, + NULL); + ucs_fatal("rc mlx5: unsupported outstanding WQE opcode 0x%x size %zu: %s", + uct_rc_mlx5_wqe_opcode(ctrl), wqe_size, wqe_dump); +} + +static uint32_t uct_ib_mlx5_wqe_num_packets( + uct_ib_iface_t *iface, const uct_ib_mlx5_txwq_t *txwq, + const struct mlx5_wqe_ctrl_seg *ctrl, size_t wqe_size) +{ + uint8_t opcode = uct_rc_mlx5_wqe_opcode(ctrl); + const struct mlx5_wqe_inl_data_seg *inl; + size_t inline_length, inline_wqe_size; + + switch (opcode) { + case MLX5_OPCODE_NOP: + return 0; + case MLX5_OPCODE_SEND: + inl = uct_rc_mlx5_wqe_get_inline_seg(txwq, ctrl, wqe_size, + &inline_length); + if (inl != NULL) { + inline_wqe_size = sizeof(*ctrl) + + ucs_align_up_pow2(sizeof(*inl) + inline_length, + UCT_IB_MLX5_WQE_SEG_SIZE); + /* AM short is fully inline. IOV-backed AM zcopy also starts + * inline, but has trailing data segments we do not support. */ + if (wqe_size == inline_wqe_size) { + return 1; + } + } + + /* Fall through */ + default: + uct_rc_mlx5_wqe_unsupported(iface, txwq, ctrl, wqe_size); + } +} + +static uint32_t uct_rc_mlx5_txwq_outstanding_num_packets( + uct_ib_iface_t *iface, uct_ib_mlx5_txwq_t *txwq, uint16_t start_ci, + uint16_t end_ci) +{ + const struct mlx5_wqe_ctrl_seg *ctrl; + uint32_t num_packets = 0; + uint16_t ci; + size_t wqe_size; + + for (ci = start_ci; ci != end_ci; + ci = uct_ib_mlx5_txwq_next_wqe_index(ci, wqe_size)) { + ctrl = uct_ib_mlx5_txwq_get_wqe(txwq, ci); + wqe_size = uct_ib_mlx5_wqe_size(ctrl); + + num_packets += uct_ib_mlx5_wqe_num_packets(iface, txwq, ctrl, wqe_size); + } + + return num_packets; +} + +static ucs_status_t uct_rc_mlx5_ep_outstanding_purge_check_params( + const uct_ep_outstanding_purge_params_t *params) +{ + if (params == NULL) { + ucs_error("rc mlx5: outstanding purge parameters are NULL"); + return UCS_ERR_INVALID_PARAM; + } + + if (!(params->field_mask & UCT_EP_OUTSTANDING_FIELD_RX_TOKEN)) { + ucs_error("rc mlx5: rx token is not set"); + return UCS_ERR_INVALID_PARAM; + } + + ucs_assert(params->rx_token != NULL); + + if (!(params->field_mask & UCT_EP_OUTSTANDING_FIELD_CB) || + (params->cb == NULL)) { + ucs_error("rc mlx5: callback is not set or is NULL"); + return UCS_ERR_INVALID_PARAM; + } + + return UCS_OK; +} + +static void +uct_rc_mlx5_ep_purge_flushes(uct_rc_mlx5_base_ep_t *ep, uint16_t ci) +{ + uct_rc_iface_send_op_t *op; + + ucs_queue_for_each_extract( + op, &ep->super.txqp.outstanding, queue, + (op->handler == uct_rc_ep_flush_op_completion_handler) && + UCS_CIRCULAR_COMPARE16(op->sn, <=, ci)) { + op->flags &= ~UCT_RC_IFACE_SEND_OP_FLAG_INUSE; + uct_invoke_completion(op->user_comp, UCS_ERR_CANCELED); + ucs_mpool_put(op); + } +} + +ucs_status_t uct_rc_mlx5_ep_outstanding_purge( + uct_ep_h tl_ep, const uct_ep_outstanding_purge_params_t *params) +{ + UCT_RC_MLX5_BASE_EP_DECL(tl_ep, iface, ep); + uct_ib_mlx5_txwq_t *txwq = &ep->tx.wq; + const uct_rc_mlx5_rx_token_t *rx_token; + const struct mlx5_wqe_ctrl_seg *ctrl; + uint8_t callback_data[UCT_IB_MLX5_MAX_SEND_WQE_SIZE]; + uct_ep_op_info_t info; + uint16_t ci, end_ci, start_ci; + uint32_t first_failed_psn, wqe_first_psn, receiver_next_psn, psn_diff; + uint32_t num_outstanding_packets, num_packets; + size_t wqe_size; + void *callback_arg; + ucs_status_t status; + + status = uct_rc_mlx5_ep_outstanding_purge_check_params(params); + if (status != UCS_OK) { + return status; + } + + callback_arg = (params->field_mask & UCT_EP_OUTSTANDING_FIELD_ARG) ? + params->arg : NULL; + + ucs_assertv_always(ep->err_handler_inprogress, + "ep %p is not in deferred error handling", ep); + + end_ci = txwq->sw_pi; + ctrl = uct_ib_mlx5_txwq_get_wqe(txwq, txwq->ft_ci); + start_ci = uct_ib_mlx5_txwq_next_wqe_index( + txwq->ft_ci, uct_ib_mlx5_wqe_size(ctrl)); + + if (start_ci == end_ci) { + goto out; + } + + num_outstanding_packets = uct_rc_mlx5_txwq_outstanding_num_packets( + &iface->super.super, txwq, start_ci, end_ci); + + rx_token = params->rx_token; + receiver_next_psn = ntohl(*rx_token) & UCT_IB_MLX5_PSN_MASK; + first_failed_psn = (uct_ib_mlx5_txwq_get_next_wqe_psn(txwq) - + num_outstanding_packets) & UCT_IB_MLX5_PSN_MASK; + psn_diff = (receiver_next_psn - first_failed_psn) & + UCT_IB_MLX5_PSN_MASK; + if (psn_diff > num_outstanding_packets) { + return UCS_ERR_INVALID_PARAM; + } + + wqe_first_psn = first_failed_psn; + for (ci = start_ci; ci != end_ci; + ci = uct_ib_mlx5_txwq_next_wqe_index(ci, wqe_size)) { + ctrl = uct_ib_mlx5_txwq_get_wqe(txwq, ci); + wqe_size = uct_ib_mlx5_wqe_size(ctrl); + num_packets = uct_ib_mlx5_wqe_num_packets(&iface->super.super, txwq, + ctrl, wqe_size); + if ((num_packets != 0) && + !uct_ib_mlx5_wqe_is_delivered(wqe_first_psn, receiver_next_psn, + num_packets)) { + status = uct_rc_mlx5_op_info_fill_am( + txwq, ctrl, wqe_size, callback_data, &info); + if (status == UCS_OK) { + params->cb(&info, callback_arg); + } else if (status != UCS_ERR_NO_ELEM) { + ucs_fatal("rc mlx5: failed to parse outstanding WQE at ci %u", + ci); + } + } + + wqe_first_psn = (wqe_first_psn + num_packets) & + UCT_IB_MLX5_PSN_MASK; + + /* Complete flushes after their WQE, before later AM purge callbacks. */ + uct_rc_mlx5_ep_purge_flushes(ep, ci); + } + +out: + uct_rc_mlx5_ep_update_tx_qp_res(ep, txwq->prev_sw_pi); + txwq->ft_ci = txwq->prev_sw_pi; + return UCS_OK; +} + ucs_status_t uct_rc_mlx5_base_ep_fc_ctrl(uct_ep_t *tl_ep, unsigned op, uct_rc_pending_req_t *req) { diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c index b611da3cd3d..8f2c00d3466 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c @@ -1164,6 +1164,10 @@ uct_rc_mlx5_iface_query_v2(uct_iface_h tl_iface, md = uct_ib_mlx5_iface_md(ucs_derived_of(tl_iface, uct_ib_iface_t)); if (md->flags & UCT_IB_MLX5_MD_FLAG_DEVX) { + if (iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_CAP_FLAGS) { + iface_attr->cap.flags |= UCT_IFACE_FLAG_V2_QUERY_TOKEN; + } + if (iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_RX_TOKEN) { status = uct_rc_mlx5_iface_query_rx_token(tl_iface, iface_attr); if (status != UCS_OK) { @@ -1239,7 +1243,7 @@ static uct_rc_iface_ops_t uct_rc_mlx5_iface_ops = { .ep_is_connected = uct_rc_mlx5_base_ep_is_connected, .ep_get_device_ep = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported, .ep_put_sgl_zcopy = uct_rc_mlx5_ep_put_sgl_zcopy, - .ep_outstanding_purge = uct_ib_mlx5_ext_ep_outstanding_purge + .ep_outstanding_purge = uct_rc_mlx5_ep_outstanding_purge }, .create_cq = uct_rc_mlx5_iface_common_create_cq, .destroy_cq = uct_rc_mlx5_iface_common_destroy_cq, diff --git a/test/gtest/Makefile.am b/test/gtest/Makefile.am index 2638992b103..c9ed084dfc1 100644 --- a/test/gtest/Makefile.am +++ b/test/gtest/Makefile.am @@ -262,7 +262,6 @@ gtest_SOURCES += \ endif if HAVE_MLX5_DV gtest_SOURCES += \ - uct/ib/test_ib_mlx5_ext.cc \ uct/ib/test_cqe_zipping.cc endif if HAVE_EFA diff --git a/test/gtest/uct/ib/test_ib_mlx5_ext.cc b/test/gtest/uct/ib/test_ib_mlx5_ext.cc deleted file mode 100644 index 0f747b40b45..00000000000 --- a/test/gtest/uct/ib/test_ib_mlx5_ext.cc +++ /dev/null @@ -1,125 +0,0 @@ -/** - * Copyright (c) NVIDIA CORPORATION & AFFILIATES, 2026. ALL RIGHTS RESERVED. - * - * See file LICENSE for terms. - */ - -#include "test_rc.h" - -extern "C" { -#include - -extern ucs_list_link_t uct_ib_mlx5_ext_plugins; -} - - -class test_uct_ib_mlx5_ext_rc : public test_rc { -public: - void init() override - { - save_plugins(); - test_rc::init(); - } - - void cleanup() override - { - uct_ib_mlx5_ext_cleanup(); - uct_test::cleanup(); - restore_plugins(); - } - -protected: - static uint64_t tx_token() - { - return 0x1234; - } - - static uint64_t rx_token() - { - return tx_token() + 1; - } - - void save_plugins() - { - if (ucs_list_is_empty(&uct_ib_mlx5_ext_plugins)) { - ucs_list_head_init(&m_saved_plugins); - } else { - ucs_list_replace(&uct_ib_mlx5_ext_plugins, &m_saved_plugins); - ucs_list_head_init(&uct_ib_mlx5_ext_plugins); - } - } - - void restore_plugins() - { - if (!ucs_list_is_empty(&m_saved_plugins)) { - ucs_list_replace(&m_saved_plugins, &uct_ib_mlx5_ext_plugins); - ucs_list_head_init(&m_saved_plugins); - } - } - - static ucs_status_t - purge_fail(uct_ep_h, const uct_ep_outstanding_purge_params_t*) - { - return UCS_ERR_UNSUPPORTED; - } - - static ucs_status_t - purge(uct_ep_h, const uct_ep_outstanding_purge_params_t *params) - { - uct_ep_op_info_t op_info = {}; - - EXPECT_EQ(rx_token(), *static_cast(params->rx_token)); - params->cb(&op_info, params->arg); - return UCS_OK; - } - - static void purge_cb(const uct_ep_op_info_t*, void *arg) - { - *static_cast(arg) = true; - } - - static void - register_plugin(const char *name, - uct_ep_outstanding_purge_func_t purge_cb = NULL) - { - uct_ib_mlx5_ext_ops_t ops = {}; - - ucs_strncpy_zero(ops.name, name, sizeof(ops.name)); - ops.ep_outstanding_purge = purge_cb; - ASSERT_UCS_OK(uct_ib_mlx5_ext_register(&ops)); - } - -private: - ucs_list_link_t m_saved_plugins = UCS_LIST_INITIALIZER(&m_saved_plugins, - &m_saved_plugins); -}; - - -UCS_TEST_P(test_uct_ib_mlx5_ext_rc, ep_outstanding_purge) -{ - uint64_t rx_token_value = rx_token(); - bool callback_invoked = false; - uct_ep_outstanding_purge_params_t params = {}; - - { - scoped_log_handler wrap_err(wrap_errors_logger); - EXPECT_EQ(UCS_ERR_INVALID_PARAM, - uct_ep_outstanding_purge(m_e1->ep(0), ¶ms)); - } - - register_plugin("stub"); - register_plugin("fail", purge_fail); - register_plugin("token", purge); - - params.field_mask = UCT_EP_OUTSTANDING_FIELD_RX_TOKEN | - UCT_EP_OUTSTANDING_FIELD_CB | - UCT_EP_OUTSTANDING_FIELD_ARG; - params.rx_token = &rx_token_value; - params.cb = purge_cb; - params.arg = &callback_invoked; - - ASSERT_UCS_OK(uct_ep_outstanding_purge(m_e1->ep(0), ¶ms)); - EXPECT_TRUE(callback_invoked); -} - -_UCT_INSTANTIATE_TEST_CASE(test_uct_ib_mlx5_ext_rc, rc_mlx5) diff --git a/test/gtest/uct/test_peer_failure.cc b/test/gtest/uct/test_peer_failure.cc index 75fb0eeaf39..397f97dfd4d 100644 --- a/test/gtest/uct/test_peer_failure.cc +++ b/test/gtest/uct/test_peer_failure.cc @@ -375,9 +375,15 @@ class test_uct_purge_outstanding : public uct_test { protected: struct purge_ctx { test_uct_purge_outstanding *self; - uct_completion_t op_comp; + uct_completion_t comp; uint32_t num_ops_purged; + uint32_t num_ops_purged_at_completion; + uint32_t num_ops_posted_after_flush; uint32_t num_flush_purged; + unsigned num_completions; + uint8_t am_short_id; + uint64_t am_short_header; + std::vector am_short_payload; }; using send_func_t = @@ -407,19 +413,18 @@ class test_uct_purge_outstanding : public uct_test { return status; } - static bool post_flush(uct_ep_h ep, uct_completion_t *comp) + static void post_flush(uct_ep_h ep, uct_completion_t *comp) { ucs_status_t status; ++comp->count; status = uct_ep_flush(ep, 0, comp); if (status == UCS_INPROGRESS) { - return true; + return; } --comp->count; ASSERT_UCS_OK(status); - return false; } static uint32_t post_until_error(uct_ep_h ep, uct_completion_t *comp, @@ -446,6 +451,9 @@ class test_uct_purge_outstanding : public uct_test { void validate_op(const uct_ep_op_info_t *info, purge_ctx *ctx) { switch (info->operation) { + case UCT_EP_OP_AM_SHORT: + validate_am_short(info, ctx); + return; case UCT_EP_OP_FLUSH: validate_flush(info, ctx); return; @@ -456,6 +464,26 @@ class test_uct_purge_outstanding : public uct_test { } } + void validate_am_short(const uct_ep_op_info_t *info, purge_ctx *ctx) + { + const uint16_t required_fields = + UCT_EP_OP_INFO_AM_FIELD_AM_ID | + UCT_EP_OP_INFO_AM_FIELD_HEADER_VALUE | + UCT_EP_OP_INFO_AM_FIELD_PAYLOAD_DATA; + + ASSERT_TRUE(info->field_mask & UCT_EP_OP_INFO_FIELD_AM); + ASSERT_TRUE(ucs_test_all_flags(info->am.field_mask, required_fields)); + EXPECT_FALSE(info->field_mask & UCT_EP_OP_INFO_FIELD_COMP); + EXPECT_EQ(ctx->am_short_id, info->am.am_id); + EXPECT_EQ(ctx->am_short_header, info->am.header.value); + ASSERT_EQ(ctx->am_short_payload.size(), info->am.payload.data.length); + ASSERT_NE(nullptr, info->am.payload.data.buffer); + EXPECT_EQ(0, memcmp(ctx->am_short_payload.data(), + info->am.payload.data.buffer, + ctx->am_short_payload.size())); + ++ctx->num_ops_purged; + } + static void validate_flush(const uct_ep_op_info_t *info, purge_ctx *ctx) { const uint64_t required_fields = UCT_EP_OP_INFO_FIELD_COMP | @@ -465,7 +493,7 @@ class test_uct_purge_outstanding : public uct_test { ASSERT_TRUE(info->flush.field_mask & UCT_EP_OP_INFO_FLUSH_FIELD_FLAGS); EXPECT_EQ(0u, info->flush.flags); - EXPECT_EQ(&ctx->op_comp, info->comp); + EXPECT_EQ(&ctx->comp, info->comp); ASSERT_EQ(0u, ctx->num_flush_purged); ++ctx->num_flush_purged; } @@ -493,8 +521,17 @@ class test_uct_purge_outstanding : public uct_test { ctx->self->validate_op(info, ctx); } - static void completion_cb(uct_completion_t*) + static void completion_cb(uct_completion_t *comp) + { + purge_ctx *ctx = ucs_container_of(comp, purge_ctx, comp); + + ctx->num_ops_purged_at_completion = ctx->num_ops_purged; + ++ctx->num_completions; + } + + static ucs_status_t am_handler(void*, void*, size_t, unsigned) { + return UCS_OK; } void purge_outstanding(purge_ctx *ctx) @@ -533,48 +570,58 @@ class test_uct_purge_outstanding : public uct_test { &purge_params)); } - void test_purge_outstanding(const send_func_t &send_func) + void test_purge_outstanding(const send_func_t &send_func, + purge_ctx &ctx) { + static constexpr uint32_t NUM_MSG_BEFORE_INVALIDATE = 2; uct_ep_invalidate_params_t invalidate_params = {}; - purge_ctx ctx = { - this, {completion_cb, 0, UCS_OK}, 0, 0}; - uint32_t num_posted; - bool flush_outstanding; + uint32_t num_posted = 0; + uint32_t num_ops_purged; + unsigned num_outstanding, num_completions; ucs_status_t status; - status = post_op(m_sender->ep(0), &ctx.op_comp, send_func); - ASSERT_UCS_OK_OR_INPROGRESS(status); - num_posted = 1; - flush(); - - status = post_op(m_sender->ep(0), &ctx.op_comp, send_func); - if (UCS_STATUS_IS_ERR(status)) { - UCS_TEST_ABORT("failed to post operation before invalidation: " - << ucs_status_string(status)); + for (; num_posted < NUM_MSG_BEFORE_INVALIDATE; ++num_posted) { + status = post_op(m_sender->ep(0), &ctx.comp, send_func); + ASSERT_UCS_OK_OR_INPROGRESS(status); } - ++num_posted; - flush_outstanding = post_flush(m_sender->ep(0), &ctx.op_comp); ASSERT_UCS_OK(uct_ep_invalidate(m_sender->ep(0), &invalidate_params)); - num_posted += post_until_error(m_sender->ep(0), &ctx.op_comp, - send_func); + /* Make flush wait for failed work even if it reuses a signaled WQE. */ + status = post_op(m_sender->ep(0), &ctx.comp, send_func); + ASSERT_UCS_OK_OR_INPROGRESS(status); + ++num_posted; + + post_flush(m_sender->ep(0), &ctx.comp); + ctx.num_ops_posted_after_flush = post_until_error( + m_sender->ep(0), &ctx.comp, send_func); + num_posted += ctx.num_ops_posted_after_flush; wait_for_flag(&m_err_count); ASSERT_EQ(1u, m_err_count); + num_outstanding = ctx.comp.count; + num_completions = ctx.num_completions; purge_outstanding(&ctx); EXPECT_GT(ctx.num_ops_purged, 0u); - EXPECT_LT(ctx.num_ops_purged, num_posted); - EXPECT_EQ(unsigned(flush_outstanding), ctx.num_flush_purged); + EXPECT_LE(ctx.num_ops_purged, num_posted); + EXPECT_EQ(0u, ctx.num_flush_purged); - wait_for_value(&ctx.op_comp.count, 0, true); - if (flush_outstanding) { - EXPECT_EQ(UCS_ERR_CANCELED, ctx.op_comp.status); + wait_for_value(&ctx.comp.count, 0, true); + if (num_outstanding != 0) { + EXPECT_EQ(UCS_ERR_CANCELED, ctx.comp.status); + ++num_completions; } + EXPECT_EQ(num_completions, ctx.num_completions); + + num_ops_purged = ctx.num_ops_purged; + purge_outstanding(&ctx); + EXPECT_EQ(num_ops_purged, ctx.num_ops_purged); + EXPECT_EQ(num_completions, ctx.num_completions); + flush(); - EXPECT_EQ(0, ctx.op_comp.count); + EXPECT_EQ(0, ctx.comp.count); } entity *m_sender; @@ -582,6 +629,29 @@ class test_uct_purge_outstanding : public uct_test { unsigned m_err_count = 0; }; +UCS_TEST_SKIP_COND_P(test_uct_purge_outstanding, am_short, + !check_caps(UCT_IFACE_FLAG_AM_SHORT)) +{ + purge_ctx ctx = {this, {completion_cb, 0, UCS_OK}, 0, 0, 0, 0, 0, + 3, 0x0123456789abcdefull, {1, 2, 3, 4}}; + + ASSERT_UCS_OK(uct_iface_set_am_handler(m_receiver->iface(), ctx.am_short_id, + am_handler, NULL, 0)); + + test_purge_outstanding([&ctx](uct_ep_h ep, uct_completion_t*) { + return uct_ep_am_short(ep, ctx.am_short_id, ctx.am_short_header, + ctx.am_short_payload.data(), + ctx.am_short_payload.size()); + }, ctx); + + EXPECT_GT(ctx.num_ops_purged_at_completion, 0u); + EXPECT_GT(ctx.num_ops_posted_after_flush, 0u); + EXPECT_EQ(ctx.num_ops_purged, ctx.num_ops_purged_at_completion + + ctx.num_ops_posted_after_flush); +} + +UCT_INSTANTIATE_TEST_CASE(test_uct_purge_outstanding) + class test_uct_peer_failure_multiple : public test_uct_peer_failure { public: