Merge pull request #10646 from tvegas1/srd_gtest_ucp

GTEST/UCP: Enable EFA SRD support

Author: brminich

buildlib/pr/efa.yml

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

contrib/test_efa.sh

@@ -103,12 +103,17 @@ 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*'
+    # TODO: Add when ibmock supports missing pre-posted RX buffers
+    IBMOCK_FILTER="srd/uct_p2p_am_misc.no_rx_buffs*:srd/test_uct_peer_failure.purge_failed_peer*"
+    IBMOCK_FILTER="$IBMOCK_FILTER:srd/test_ucp_stream_many2one.send_worker_poll*"
+    IBMOCK_FILTER="$IBMOCK_FILTER:srd/test_ucp_peer_failure.*"
+    IBMOCK_FILTER="$IBMOCK_FILTER:srd/test_ucp_perf.envelope/*"
+
     # Try the faster approach before valgrind
     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
+        GTEST_FILTER=*ud*:srd/*:-*test_uct_perf.envelope*:*test_ucp*:$IBMOCK_FILTER
 }
 
 test_ucx_rpm() {

test/gtest/ucp/test_ucp_am.cc

@@ -934,8 +934,8 @@ UCS_TEST_P(test_ucp_am_nbx, rx_am_mpools,
     set_am_data_handler(receiver(), TEST_AM_NBX_ID, am_data_hold_cb, &rx_data,
                         UCP_AM_FLAG_PERSISTENT_DATA);
 
-    static const std::string ib_tls[] = { "dc_x", "rc_v", "rc_x", "ud_v",
-                                          "ud_x", "ib" };
+    static const std::string ib_tls[] = {"dc_x", "rc_v", "rc_x", "ud_v",
+                                         "ud_x", "srd", "ib"};
 
     // UCP takes desc from mpool only for data arrived as inlined from UCT.
     // Typically, with IB, data is inlined up to 32 bytes, so use smaller range

test/gtest/ucp/test_ucp_atomic.cc

@@ -250,7 +250,8 @@ class test_ucp_atomic : public test_ucp_memheap {
                     continue;
                 }
 
-                static const std::string tls[] = { "ud_v", "ud_x", "rc_v", "tcp" };
+                static const std::string tls[] = {"ud_v", "ud_x", "rc_v", "tcp",
+                                                  "srd"};
                 /* Target memory type atomics emulation not supported yet */
                 if (((atomic_mode == UCP_ATOMIC_MODE_CPU) ||
                      has_any_transport(tls, ucs_static_array_size(tls))) &&

test/gtest/ucp/test_ucp_context.cc

@@ -68,6 +68,7 @@ UCS_TEST_P(test_ucp_aliases, aliases) {
 UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_aliases, rcv, "rc_v")
 UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_aliases, rcx, "rc_x")
 UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_aliases, ud, "ud")
+UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_aliases, srd, "srd")
 UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_aliases, ud_mlx5, "ud_mlx5")
 UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_aliases, ugni, "ugni")
 UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_aliases, shm, "shm")

test/gtest/ucp/test_ucp_perf.cc

@@ -334,9 +334,9 @@ UCS_TEST_SKIP_COND_P(test_ucp_perf, envelope, has_transport("self"))
     test_spec test  = tests[get_variant_value(VARIANT_TEST_TYPE)];
 
     if (ucs::is_aws() && (test.wait_mode == UCX_PERF_WAIT_MODE_SLEEP) &&
-        has_transport("ud_v")) {
+        (has_transport("ud_v") || has_transport("srd"))) {
         // TODO support wakeup in UD transport without requiring IBV_SEND_SOLICITED
-        UCS_TEST_SKIP_R("wait mode sleep on EFA not available on UD");
+        UCS_TEST_SKIP_R("wait mode sleep on EFA not available");
     }
 
     if (has_transport("tcp")) {

test/gtest/ucp/test_ucp_request.cc

@@ -741,8 +741,18 @@ class test_proto_reset_atp : public test_proto_reset {
                     UCP_PROTO_RNDV_PUT_STAGE_FENCED_ATP);
         });
 
-        restart(sender().ep());
+
+       /*
+        * Restore the pending_add callbacks, as some transport need to be able
+        * to add ep flush request to the pending queue. For instance RDNV put
+        * and fence scheme will block ep_flush on SRD transport.
+        *
+        * Purge must be done with test callbacks first.
+        */
+        ucp_ep_purge_lanes(sender().ep(), purge_enqueue_cb, this);
         restore_uct_cbs();
+
+        restart(sender().ep());
     }
 
     static unsigned m_atp_count;

test/gtest/ucp/test_ucp_rma.cc

@@ -585,6 +585,14 @@ class test_ucp_ep_based_fence : public test_ucp_rma {
         ucs::handle<ucp_rkey_h> rkey;
         rbuf.rkey(sender(), rkey);
 
+        /*
+         * Some transports might need some bidirectional progress to be able
+         * to transmit without using their pending queue.
+         */
+        do_rma_op(op, sbuf.ptr(), sizeof(uint32_t), rbuf.ptr(), rkey);
+        flush_workers();
+        flush_sender_cleanup();
+
         if (op == OP_ATOMIC) {
             (this->*fence_func)(op, sbuf.ptr(), sizeof(uint32_t), rbuf.ptr(),
                                 rkey);
@@ -599,12 +607,7 @@ class test_ucp_ep_based_fence : public test_ucp_rma {
         flush_workers();
     }
 
-    void do_rma_op_with_fence_before(op_type_t op, void *sbuf, size_t size,
-                                     void *target, ucp_rkey_h rkey) {
-        do_fence();
-        do_rma_op(op, sbuf, size, target, rkey);
-
-        flush_worker(sender());
+    void flush_sender_cleanup() {
         /*
          * flush_worker() doesn't reset unflushed_lanes yet (planned).
          * Manually clear unflushed_lanes to simulate a fence-before-op scenario
@@ -615,6 +618,15 @@ class test_ucp_ep_based_fence : public test_ucp_rma {
          */
         sender().ep()->ext->unflushed_lanes = 0;
     }
+
+    void do_rma_op_with_fence_before(op_type_t op, void *sbuf, size_t size,
+                                     void *target, ucp_rkey_h rkey) {
+        do_fence();
+        do_rma_op(op, sbuf, size, target, rkey);
+
+        flush_worker(sender());
+        flush_sender_cleanup();
+    }
 private:
     static constexpr size_t TEST_BUF_SIZE = 1000000;
 };

test/gtest/ucp/test_ucp_sockaddr.cc

@@ -2837,6 +2837,7 @@ UCS_TEST_SKIP_COND_P(test_ucp_sockaddr_protocols, am_rndv_reset,
     UCP_INSTANTIATE_TEST_CASE_TLS_GPU_AWARE(_test_case, udx, "ud_x") \
     UCP_INSTANTIATE_TEST_CASE_TLS_GPU_AWARE(_test_case, rc, "rc_v") \
     UCP_INSTANTIATE_TEST_CASE_TLS_GPU_AWARE(_test_case, rcx, "rc_x") \
+    UCP_INSTANTIATE_TEST_CASE_TLS_GPU_AWARE(_test_case, srd, "srd") \
     UCP_INSTANTIATE_TEST_CASE_TLS_GPU_AWARE(_test_case, ib, "ib") \
     UCP_INSTANTIATE_TEST_CASE_TLS_GPU_AWARE(_test_case, tcp, "tcp") \
     UCP_INSTANTIATE_TEST_CASE_TLS(_test_case, all, "all")

test/gtest/ucp/test_ucp_tag_mem_type.cc

@@ -33,8 +33,8 @@ class test_ucp_tag_mem_type: public test_ucp_tag {
         int variant_flags = get_variant_value() / m_mem_type_pairs.size();
 
         if (variant_flags & VARIANT_GDR_OFF) {
-            if (!has_any_transport(
-                        {"dc_x", "ud_v", "ud_x", "rc_v", "rc_x", "ib"})) {
+            if (!has_any_transport({"dc_x", "ud_v", "ud_x", "rc_v", "rc_x",
+                                    "srd", "ib"})) {
                 UCS_TEST_SKIP_R("No GPU direct RDMA");
             }