Merge pull request #6376 from Artemy-Mellanox/topic/lag-6

UCT/DC: DCI pool per LAG port

Author: yosefe

buildlib/io_demo/az-stage-io-demo.yaml

@@ -3,6 +3,8 @@ parameters:
   default: 'test'
 - name: iodemo_args
   default: ''
+- name: iodemo_tls
+  default: 'rc_x'
 
 steps:
 - bash: |
@@ -31,6 +33,7 @@ steps:
     mkdir -p $(workspace)/${{ parameters.name }}
     # set UCX environment variables
     export UCX_NET_DEVICES=$(ibdev2netdev | sed -ne 's/\(\w*\) port \([0-9]\) ==> '${roce_iface}' .*/\1:\2/p')
+    export UCX_TLS=${{ parameters.iodemo_tls }}
     export LD_LIBRARY_PATH=$(workspace)/install/lib:$LD_LIBRARY_PATH
     $(workspace)/test/apps/iodemo/run_io_demo.sh \
         -H $(agent_hosts) \

src/uct/base/uct_iface.h

@@ -439,7 +439,7 @@ uct_pending_req_priv_arb_elem(uct_pending_req_t *req)
 /**
  * Add a pending request to the head of group in arbiter.
  */
-#define uct_pending_req_arb_group_push_head(_arbiter, _arbiter_group, _req) \
+#define uct_pending_req_arb_group_push_head(_arbiter_group, _req) \
     do { \
         ucs_arbiter_elem_init(uct_pending_req_priv_arb_elem(_req)); \
         ucs_arbiter_group_push_head_elem_always(_arbiter_group, \

src/uct/ib/base/ib_iface.c

@@ -662,9 +662,8 @@ void uct_ib_iface_fill_ah_attr_from_gid_lid(uct_ib_iface_t *iface, uint16_t lid,
     if (uct_ib_iface_is_roce(iface)) {
         ah_attr->dlid          = UCT_IB_ROCE_UDP_SRC_PORT_BASE |
                                  (iface->config.roce_path_factor * path_index);
-        /* Workaround rdma-core issue of calling rand() which affects global
-         * random state in glibc */
-        ah_attr->grh.flow_label = 1;
+        /* Workaround rdma-core flow label to udp sport conversion */
+        ah_attr->grh.flow_label = ~(iface->config.roce_path_factor * path_index);
     } else {
         /* TODO iface->path_bits should be removed and replaced by path_index */
         path_bits              = iface->path_bits[path_index %

src/uct/ib/dc/dc_mlx5.c

@@ -38,8 +38,10 @@ struct ibv_ravh {
 
 #define UCT_DC_MLX5_IFACE_MAX_USER_DCIS 15
 #define UCT_DC_MLX5_KEEPALIVE_NUM_DCIS  1
-#define UCT_DC_MLX5_IFACE_MAX_DCIS      (UCT_DC_MLX5_IFACE_MAX_USER_DCIS + \
-                                         UCT_DC_MLX5_KEEPALIVE_NUM_DCIS)
+#define UCT_DC_MLX5_IFACE_MAX_DCI_POOLS 8
+#define UCT_DC_MLX5_IFACE_MAX_DCIS      ((UCT_DC_MLX5_IFACE_MAX_USER_DCIS * \
+                                          UCT_DC_MLX5_IFACE_MAX_DCI_POOLS) + \
+                                          UCT_DC_MLX5_KEEPALIVE_NUM_DCIS)
 
 #define UCT_DC_MLX5_IFACE_ADDR_TM_ENABLED(_addr) \
     (!!((_addr)->flags & UCT_DC_MLX5_IFACE_ADDR_HW_TM))
@@ -130,7 +132,7 @@ typedef struct uct_dc_mlx5_iface_config {
 
 typedef void (*uct_dc_dci_handle_failure_func_t)(uct_dc_mlx5_iface_t *iface,
                                                  struct mlx5_cqe64 *cqe,
-                                                 uint8_t dci,
+                                                 uint8_t dci_index,
                                                  ucs_status_t status);
 
 
@@ -147,6 +149,7 @@ typedef struct uct_dc_dci {
                                                 processed. Better have dci num
                                                 groups scheduled than ep num. */
     };
+    uint8_t                       pool_index; /* DCI pool index. */
 #if UCS_ENABLE_ASSERT
     uint8_t                       flags; /* debug state, @ref uct_dc_dci_state_t */
 #endif
@@ -175,22 +178,32 @@ typedef struct uct_dc_fc_request {
 
 KHASH_MAP_INIT_INT64(uct_dc_mlx5_fc_hash, uint64_t);
 
+
+typedef struct {
+    uint8_t       stack_top;                               /* dci stack top */
+    uint8_t       stack[UCT_DC_MLX5_IFACE_MAX_USER_DCIS];  /* LIFO of indexes of available dcis */
+    ucs_arbiter_t arbiter;                                 /* queue of requests
+                                                              waiting for DCI */
+} uct_dc_mlx5_dci_pool_t;
+
+
 struct uct_dc_mlx5_iface {
     uct_rc_mlx5_iface_common_t    super;
     struct {
         /* Array of dcis */
         uct_dc_dci_t              dcis[UCT_DC_MLX5_IFACE_MAX_DCIS];
 
         uint8_t                   ndci;                        /* Number of DCIs */
-        uct_dc_tx_policy_t        policy;                      /* dci selection algorithm */
-        int16_t                   available_quota;             /* if available tx is lower, let
-                                                                  another endpoint use the dci */
 
         /* LIFO is only relevant for dcs allocation policy */
-        uint8_t                   stack_top;                   /* dci stack top */
-        uint8_t                   dcis_stack[UCT_DC_MLX5_IFACE_MAX_DCIS];  /* LIFO of indexes of available dcis */
+        uct_dc_mlx5_dci_pool_t    dci_pool[UCT_DC_MLX5_IFACE_MAX_DCI_POOLS];
+        uint8_t                   num_dci_pools;
 
-        ucs_arbiter_t             dci_arbiter;
+        uint8_t                   policy;                      /* dci selection algorithm */
+        int16_t                   available_quota;             /* if available tx is lower, let
+                                                                  another endpoint use the dci */
+        /* DCI max elements */
+        unsigned                  bb_max;
 
         /* Used to send grant messages for all peers */
         uct_dc_mlx5_ep_t          *fc_ep;
@@ -216,6 +229,8 @@ struct uct_dc_mlx5_iface {
     /* iface flags, see uct_dc_mlx5_iface_flags_t */
     uint8_t                       flags;
 
+    uint8_t                       keepalive_dci;
+
     uct_ud_mlx5_iface_common_t    ud_common;
 };
 
@@ -261,7 +276,9 @@ void uct_dc_mlx5_iface_set_ep_failed(uct_dc_mlx5_iface_t *iface,
                                      uct_ib_mlx5_txwq_t *txwq,
                                      ucs_status_t ep_status);
 
-void uct_dc_mlx5_iface_reset_dci(uct_dc_mlx5_iface_t *iface, uint8_t dci);
+ucs_status_t uct_dc_mlx5_iface_create_dcis(uct_dc_mlx5_iface_t *iface);
+
+void uct_dc_mlx5_iface_reset_dci(uct_dc_mlx5_iface_t *iface, uint8_t dci_index);
 
 #if HAVE_DEVX
 
@@ -270,7 +287,8 @@ ucs_status_t uct_dc_mlx5_iface_devx_create_dct(uct_dc_mlx5_iface_t *iface);
 ucs_status_t uct_dc_mlx5_iface_devx_set_srq_dc_params(uct_dc_mlx5_iface_t *iface);
 
 ucs_status_t uct_dc_mlx5_iface_devx_dci_connect(uct_dc_mlx5_iface_t *iface,
-                                                uct_ib_mlx5_qp_t *qp);
+                                                uct_ib_mlx5_qp_t *qp,
+                                                uint8_t pool_index);
 
 #else
 
@@ -305,11 +323,12 @@ uct_dc_mlx5_iface_fill_ravh(struct ibv_ravh *ravh, uint32_t dct_num)
 }
 #endif
 
-static inline uint8_t
+static UCS_F_ALWAYS_INLINE uint8_t
 uct_dc_mlx5_iface_total_ndci(uct_dc_mlx5_iface_t *iface)
 {
-    return iface->tx.ndci + ((iface->flags & UCT_DC_MLX5_IFACE_FLAG_KEEPALIVE) ?
-                             UCT_DC_MLX5_KEEPALIVE_NUM_DCIS : 0);
+    return (iface->tx.ndci * iface->tx.num_dci_pools) +
+        ((iface->flags & UCT_DC_MLX5_IFACE_FLAG_KEEPALIVE) ?
+         UCT_DC_MLX5_KEEPALIVE_NUM_DCIS : 0);
 }
 
 /* TODO:
@@ -318,13 +337,20 @@ uct_dc_mlx5_iface_total_ndci(uct_dc_mlx5_iface_t *iface)
  * linear search is most probably the best way to go
  * because the number of dcis is usually small
  */
-static inline uint8_t uct_dc_mlx5_iface_dci_find(uct_dc_mlx5_iface_t *iface, uint32_t qp_num)
+static UCS_F_ALWAYS_INLINE uint8_t
+uct_dc_mlx5_iface_dci_find(uct_dc_mlx5_iface_t *iface, struct mlx5_cqe64 *cqe)
 {
-    uct_dc_dci_t *dcis = iface->tx.dcis;
-    int i, ndci = uct_dc_mlx5_iface_total_ndci(iface);
+    uint32_t qp_num;
+    int i, ndci;
 
+    if (ucs_likely(iface->flags & UCT_DC_MLX5_IFACE_FLAG_UIDX)) {
+        return cqe->srqn_uidx >> UCT_IB_UIDX_SHIFT;
+    }
+
+    qp_num = ntohl(cqe->sop_drop_qpn) & UCS_MASK(UCT_IB_QPN_ORDER);
+    ndci   = uct_dc_mlx5_iface_total_ndci(iface);
     for (i = 0; i < ndci; i++) {
-        if (dcis[i].txwq.super.qp_num == qp_num) {
+        if (iface->tx.dcis[i].txwq.super.qp_num == qp_num) {
             return i;
         }
     }
@@ -339,52 +365,55 @@ uct_dc_mlx5_iface_has_tx_resources(uct_dc_mlx5_iface_t *iface)
            (iface->super.super.tx.reads_available > 0);
 }
 
-static inline int uct_dc_mlx5_iface_dci_has_tx_resources(uct_dc_mlx5_iface_t *iface, uint8_t dci)
+static UCS_F_ALWAYS_INLINE int
+uct_dc_mlx5_iface_dci_has_tx_resources(uct_dc_mlx5_iface_t *iface,
+                                       uint8_t dci_index)
 {
-    return uct_rc_txqp_available(&iface->tx.dcis[dci].txqp) > 0;
+    return uct_rc_txqp_available(&iface->tx.dcis[dci_index].txqp) > 0;
 }
 
 /* returns pending queue of eps waiting for tx resources */
-static inline ucs_arbiter_t *uct_dc_mlx5_iface_tx_waitq(uct_dc_mlx5_iface_t *iface)
+static UCS_F_ALWAYS_INLINE ucs_arbiter_t *
+uct_dc_mlx5_iface_tx_waitq(uct_dc_mlx5_iface_t *iface)
 {
-    return &iface->tx.dci_arbiter;
+    return &iface->super.super.tx.arbiter;
 }
 
 /* returns pending queue of eps waiting for the dci allocation */
-static inline ucs_arbiter_t *uct_dc_mlx5_iface_dci_waitq(uct_dc_mlx5_iface_t *iface)
+static UCS_F_ALWAYS_INLINE ucs_arbiter_t *
+uct_dc_mlx5_iface_dci_waitq(uct_dc_mlx5_iface_t *iface, uint8_t pool_index)
 {
-    return &iface->super.super.tx.arbiter;
+    return &iface->tx.dci_pool[pool_index].arbiter;
 }
 
-static inline int
-uct_dc_mlx5_iface_dci_has_outstanding(uct_dc_mlx5_iface_t *iface, int dci)
+static UCS_F_ALWAYS_INLINE int
+uct_dc_mlx5_iface_dci_has_outstanding(uct_dc_mlx5_iface_t *iface, int dci_index)
 {
     uct_rc_txqp_t *txqp;
 
-    txqp = &iface->tx.dcis[dci].txqp;
-    return uct_rc_txqp_available(txqp) < (int16_t)iface->super.super.config.tx_qp_len;
+    txqp = &iface->tx.dcis[dci_index].txqp;
+    return uct_rc_txqp_available(txqp) < (int16_t)iface->tx.bb_max;
 }
 
-static inline ucs_status_t uct_dc_mlx5_iface_flush_dci(uct_dc_mlx5_iface_t *iface, int dci)
+static UCS_F_ALWAYS_INLINE ucs_status_t
+uct_dc_mlx5_iface_flush_dci(uct_dc_mlx5_iface_t *iface, int dci_index)
 {
 
-    if (!uct_dc_mlx5_iface_dci_has_outstanding(iface, dci)) {
+    if (!uct_dc_mlx5_iface_dci_has_outstanding(iface, dci_index)) {
         return UCS_OK;
     }
-    ucs_trace_poll("dci %d is not flushed %d/%d", dci,
-                   iface->tx.dcis[dci].txqp.available,
-                   iface->super.super.config.tx_qp_len);
-    ucs_assertv(uct_rc_txqp_unsignaled(&iface->tx.dcis[dci].txqp) == 0,
+
+    ucs_trace_poll("dci %d is not flushed %d/%d", dci_index,
+                   iface->tx.dcis[dci_index].txqp.available, iface->tx.bb_max);
+    ucs_assertv(uct_rc_txqp_unsignaled(&iface->tx.dcis[dci_index].txqp) == 0,
                 "unsignalled send is not supported!!!");
     return UCS_INPROGRESS;
 }
 
-static inline int
-uct_dc_mlx5_iface_is_dci_keepalive(uct_dc_mlx5_iface_t *iface, int dci)
+static UCS_F_ALWAYS_INLINE int
+uct_dc_mlx5_iface_is_dci_keepalive(uct_dc_mlx5_iface_t *iface, int dci_index)
 {
-    ucs_assert(dci < uct_dc_mlx5_iface_total_ndci(iface));
-
-    return dci == iface->tx.ndci;
+    return dci_index == iface->keepalive_dci;
 }
 
 #endif

src/uct/ib/dc/dc_mlx5.h

@@ -74,14 +74,19 @@ ucs_status_t uct_dc_mlx5_iface_devx_create_dct(uct_dc_mlx5_iface_t *iface)
 
 ucs_status_t
 uct_dc_mlx5_iface_devx_dci_connect(uct_dc_mlx5_iface_t *iface,
-                                   uct_ib_mlx5_qp_t *qp)
+                                   uct_ib_mlx5_qp_t *qp,
+                                   uint8_t lag_port)
 {
+    uct_ib_device_t *dev = uct_ib_iface_device(&iface->super.super.super);
+    uct_ib_mlx5_md_t *md = ucs_derived_of(iface->super.super.super.super.md,
+                                          uct_ib_mlx5_md_t);
     char in_2init[UCT_IB_MLX5DV_ST_SZ_BYTES(rst2init_qp_in)]   = {};
     char out_2init[UCT_IB_MLX5DV_ST_SZ_BYTES(rst2init_qp_out)] = {};
     char in_2rtr[UCT_IB_MLX5DV_ST_SZ_BYTES(init2rtr_qp_in)]    = {};
     char out_2rtr[UCT_IB_MLX5DV_ST_SZ_BYTES(init2rtr_qp_out)]  = {};
     char in_2rts[UCT_IB_MLX5DV_ST_SZ_BYTES(rtr2rts_qp_in)]     = {};
     char out_2rts[UCT_IB_MLX5DV_ST_SZ_BYTES(rtr2rts_qp_out)]   = {};
+    uint32_t opt_param_mask = UCT_IB_MLX5_QP_OPTPAR_RAE;
     ucs_status_t status;
     void *qpc;
 
@@ -101,7 +106,6 @@ uct_dc_mlx5_iface_devx_dci_connect(uct_dc_mlx5_iface_t *iface,
 
     UCT_IB_MLX5DV_SET(init2rtr_qp_in, in_2rtr, opcode, UCT_IB_MLX5_CMD_OP_INIT2RTR_QP);
     UCT_IB_MLX5DV_SET(init2rtr_qp_in, in_2rtr, qpn, qp->qp_num);
-    UCT_IB_MLX5DV_SET(init2rtr_qp_in, in_2rtr, opt_param_mask, 4);
 
     qpc = UCT_IB_MLX5DV_ADDR_OF(init2rtr_qp_in, in_2rtr, qpc);
     UCT_IB_MLX5DV_SET(qpc, qpc, pm_state, UCT_IB_MLX5_QPC_PM_STATE_MIGRATED);
@@ -112,11 +116,18 @@ uct_dc_mlx5_iface_devx_dci_connect(uct_dc_mlx5_iface_t *iface,
     if (uct_ib_iface_is_roce(&iface->super.super.super)) {
         UCT_IB_MLX5DV_SET(qpc, qpc, primary_address_path.eth_prio,
                           iface->super.super.super.config.sl);
+
+        if (md->flags & UCT_IB_MLX5_MD_FLAG_LAG) {
+            opt_param_mask |= UCT_IB_MLX5_QP_OPTPAR_LAG_TX_AFF;
+            UCT_IB_MLX5DV_SET(qpc, qpc, lag_tx_port_affinity,
+                              dev->first_port + lag_port);
+        }
     } else {
         UCT_IB_MLX5DV_SET(qpc, qpc, primary_address_path.sl,
                           iface->super.super.super.config.sl);
     }
 
+    UCT_IB_MLX5DV_SET(init2rtr_qp_in, in_2rtr, opt_param_mask, opt_param_mask);
     status = uct_ib_mlx5_devx_modify_qp(qp, in_2rtr, sizeof(in_2rtr),
                                         out_2rtr, sizeof(out_2rtr));
     if (status != UCS_OK) {