UCT/EFA/SRD: Add flush and fence, enable for UCT

Author: tvegas1

buildlib/pr/efa.yml

@@ -9,7 +9,7 @@ jobs:
       demands: ${{ parameters.demands }}
     displayName: ${{ parameters.name }} on ${{ parameters.container }}
     container: ${{ parameters.container }}
-    timeoutInMinutes: 90
+    timeoutInMinutes: 120
     workspace:
       clean: outputs
     steps:

contrib/test_efa.sh

@@ -103,9 +103,12 @@ run_gtests() {
         ./install/bin/ucx_perftest -l -t tag_bw
     ./install/bin/ucx_info -d
 
+    IBMOCK_FILTER='srd/uct_p2p_am_misc.no_rx_buffs*:srd/test_uct_peer_failure.purge_failed_peer*'
     # Try the faster approach before valgrind
-    make -C contrib/test/gtest test GTEST_FILTER=*ud*:*test_srd*
-    make -C contrib/test/gtest test_valgrind GTEST_FILTER=*ud*:*test_srd*:-*test_uct_perf.envelope*
+    make -C contrib/test/gtest test \
+        GTEST_FILTER=*ud*:srd/*-$IBMOCK_FILTER
+    make -C contrib/test/gtest test_valgrind \
+        GTEST_FILTER=*ud*:srd/*:-*test_uct_perf.envelope*:$IBMOCK_FILTER
 }
 
 test_ucx_rpm() {

src/ucp/core/ucp_context.c

@@ -602,13 +602,14 @@ static ucs_config_field_t ucp_config_table[] = {
    " - sm/shm  : all shared memory transports (mm, cma, knem).\n"
    " - mm      : shared memory transports - only memory mappers.\n"
    " - ugni    : ugni_smsg and ugni_rdma (uses ugni_udt for bootstrap).\n"
-   " - ib      : all infiniband transports (rc/rc_mlx5, ud/ud_mlx5, dc_mlx5).\n"
+   " - ib      : all infiniband transports (rc/rc_mlx5, ud/ud_mlx5, dc_mlx5, srd).\n"
    " - rc_v    : rc verbs (uses ud for bootstrap).\n"
    " - rc_x    : rc with accelerated verbs (uses ud_mlx5 for bootstrap).\n"
    " - rc      : rc_v and rc_x (preferably if available).\n"
    " - ud_v    : ud verbs.\n"
    " - ud_x    : ud with accelerated verbs.\n"
    " - ud      : ud_v and ud_x (preferably if available).\n"
+   " - srd     : EFA srd reliable transport.\n"
    " - dc/dc_x : dc with accelerated verbs.\n"
    " - tcp     : sockets over TCP/IP.\n"
    " - cuda    : CUDA (NVIDIA GPU) memory support.\n"
@@ -690,7 +691,8 @@ static ucp_tl_alias_t ucp_tl_aliases[] = {
   { "sm",    { "posix", "sysv", "xpmem", "knem", "cma", NULL } },
   { "shm",   { "posix", "sysv", "xpmem", "knem", "cma", NULL } },
   { "ib",    { "rc_verbs", "ud_verbs", "rc_mlx5", "ud_mlx5", "dc_mlx5",
-               "gga_mlx5", UCP_TL_AUX("ud_mlx5"), UCP_TL_AUX("ud_verbs"), NULL } },
+               "gga_mlx5", UCP_TL_AUX("ud_mlx5"), UCP_TL_AUX("ud_verbs"),
+               "srd", NULL } },
   { "ud_v",  { "ud_verbs", NULL } },
   { "ud_x",  { "ud_mlx5", NULL } },
   { "ud",    { "ud_mlx5", "ud_verbs", NULL } },

src/uct/ib/base/ib_iface.c

@@ -2104,3 +2104,24 @@ ucs_status_t uct_ib_iface_arm_cq(uct_ib_iface_t *iface,
     }
     return UCS_OK;
 }
+
+int uct_ib_iface_is_connected(uct_ib_iface_t *ib_iface,
+                              const uct_ib_address_t *ib_addr,
+                              unsigned path_index, struct ibv_ah *peer_ah)
+{
+    enum ibv_mtu path_mtu;
+    ucs_status_t status;
+    struct ibv_ah_attr ah_attr;
+    struct ibv_ah *ah;
+
+    uct_ib_iface_fill_ah_attr_from_addr(ib_iface, ib_addr, path_index,
+                                        &ah_attr, &path_mtu);
+
+    status = uct_ib_device_get_ah_cached(uct_ib_iface_device(ib_iface),
+                                         &ah_attr, &ah);
+    if (status != UCS_OK) {
+        return 0;
+    }
+
+    return ah == peer_ah;
+}

src/uct/ib/base/ib_iface.h

@@ -494,6 +494,11 @@ unsigned uct_ib_iface_address_pack_flags(uct_ib_iface_t *iface);
 size_t uct_ib_iface_address_size(uct_ib_iface_t *iface);
 
 
+int uct_ib_iface_is_connected(uct_ib_iface_t *ib_iface,
+                              const uct_ib_address_t *ib_addr,
+                              unsigned path_index, struct ibv_ah *peer_ah);
+
+
 /**
  * Pack IB address.
  *
@@ -660,16 +665,23 @@ uint16_t uct_ib_iface_resolve_remote_flid(uct_ib_iface_t *iface,
     uct_ib_iface_is_roce(_iface) ? "RoCE" : "IB"
 
 
-#define UCT_IB_IFACE_VERBS_COMPLETION_ERR(_type, _iface, _i,  _wc) \
-    ucs_fatal("%s completion[%d] with error on %s/%p: %s, vendor_err 0x%x wr_id 0x%lx", \
-              _type, _i, uct_ib_device_name(uct_ib_iface_device(_iface)), _iface, \
-              uct_ib_wc_status_str(_wc[_i].status), _wc[_i].vendor_err, \
-              _wc[_i].wr_id);
+#define UCT_IB_IFACE_VERBS_COMPLETION_MSG(_type, _iface, _i, _wc) \
+            "%s completion[%d] with error on %s/%p: %s," \
+            " vendor_err 0x%x wr_id 0x%lx", \
+            _type, _i, uct_ib_device_name(uct_ib_iface_device(_iface)), \
+            _iface, uct_ib_wc_status_str(_wc[_i].status), _wc[_i].vendor_err, \
+            _wc[_i].wr_id
+
+#define UCT_IB_IFACE_VERBS_COMPLETION_LOG(_log_lvl, _type, _iface, _i, _wc) \
+    ucs_log(_log_lvl, UCT_IB_IFACE_VERBS_COMPLETION_MSG(_type,  _iface, _i, _wc))
+
+#define UCT_IB_IFACE_VERBS_COMPLETION_FATAL(_type, _iface, _i, _wc) \
+    ucs_fatal(UCT_IB_IFACE_VERBS_COMPLETION_MSG(_type,  _iface, _i, _wc))
 
 #define UCT_IB_IFACE_VERBS_FOREACH_RXWQE(_iface, _i, _hdr, _wc, _wc_count) \
     for (_i = 0; _i < _wc_count && ({ \
         if (ucs_unlikely(_wc[_i].status != IBV_WC_SUCCESS)) { \
-            UCT_IB_IFACE_VERBS_COMPLETION_ERR("receive", _iface, _i, _wc); \
+            UCT_IB_IFACE_VERBS_COMPLETION_FATAL("receive", _iface, _i, _wc); \
         } \
         _hdr = (typeof(_hdr))uct_ib_iface_recv_desc_hdr(_iface, \
                                                       (uct_ib_iface_recv_desc_t *)(uintptr_t)_wc[_i].wr_id); \
@@ -765,4 +777,26 @@ uct_ib_fill_cq_attr(struct ibv_cq_init_attr_ex *cq_attr,
 }
 #endif /* HAVE_DECL_IBV_CREATE_CQ_EX */
 
+static UCS_F_ALWAYS_INLINE ucs_status_t
+uct_ib_wc_to_ucs_status(enum ibv_wc_status status)
+{
+    switch (status)
+    {
+    case IBV_WC_SUCCESS:
+        return UCS_OK;
+    case IBV_WC_REM_ACCESS_ERR:
+    case IBV_WC_REM_OP_ERR:
+    case IBV_WC_REM_INV_RD_REQ_ERR:
+        return UCS_ERR_CONNECTION_RESET;
+    case IBV_WC_RETRY_EXC_ERR:
+    case IBV_WC_RNR_RETRY_EXC_ERR:
+    case IBV_WC_REM_ABORT_ERR:
+        return UCS_ERR_ENDPOINT_TIMEOUT;
+    case IBV_WC_WR_FLUSH_ERR:
+        return UCS_ERR_CANCELED;
+    default:
+        return UCS_ERR_IO_ERROR;
+    }
+}
+
 #endif