UCP/RMA/FLUSH: Add correctness test

openucx · Feb 23, 2025 · e6279da · e6279da
1 parent f0aedf4
commit e6279da
Showing 1 changed file with 119 additions and 0 deletions.
diff --git a/test/gtest/ucp/test_ucp_rma.cc b/test/gtest/ucp/test_ucp_rma.cc
@@ -485,3 +485,122 @@ UCS_TEST_P(test_ucp_rma_order, put_ordering) {
 // TODO: Strong fence hangs with SW RMA emulation, because it requires progress
 // on both peers. Add other tls, when fence implementation revised
 UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_rma_order, shm_rc_dc, "self,shm,rc,dc")
+
+class test_ucp_ep_based_fence : public test_ucp_rma {
+public:
+    static void get_test_variants(std::vector<ucp_test_variant>& variants) {
+        add_variant(variants, UCP_FEATURE_RMA | UCP_FEATURE_AMO64);
+    }
+
+    virtual void init() {
+        modify_config("FENCE_MODE", "ep_based");
+        modify_config("MAX_RMA_LANES", "2");
+        test_ucp_memheap::init();
+    }
+
+    void do_fence() {
+        uint32_t worker_fence_seq_before, worker_fence_seq_after;
+        worker_fence_seq_before = sender().ep()->worker->fence_seq;
+        sender().fence();
+        worker_fence_seq_after = sender().ep()->worker->fence_seq;
+        ASSERT_EQ(worker_fence_seq_before + 1, worker_fence_seq_after);
+        ASSERT_GT(worker_fence_seq_after, sender().ep()->ext->fence_seq);
+    }
+
+    void put_nbx(void *sbuf, size_t size, uint64_t target, ucp_rkey_h rkey) {
+        ucp_request_param_t param = {0};
+        ucs_status_ptr_t sptr = ucp_put_nbx(sender().ep(), sbuf, size, target,
+                                            rkey, &param);
+        ASSERT_FALSE(UCS_PTR_IS_ERR(sptr));
+        ASSERT_NE(sender().ep()->ext->unflushed_lanes, 0);
+        request_release(sptr);
+    }
+
+    void get_nbx(void *sbuf, size_t size, uint64_t target, ucp_rkey_h rkey) {
+        ucp_request_param_t param = {0};
+
+        ucs_status_ptr_t sptr = ucp_get_nbx(sender().ep(), sbuf, size, target,
+                                            rkey, &param);
+        ASSERT_FALSE(UCS_PTR_IS_ERR(sptr));
+        ASSERT_NE(sender().ep()->ext->unflushed_lanes, 0);
+        request_release(sptr);
+    }
+
+    void atomic_op_nbx(void *sbuf, size_t size, uint64_t target, ucp_rkey_h rkey) {
+        ucp_request_param_t param = { .op_attr_mask = UCP_OP_ATTR_FIELD_DATATYPE,
+                                      .datatype     = ucp_dt_make_contig(size) };
+
+        ucs_status_ptr_t sptr = ucp_atomic_op_nbx(sender().ep(), UCP_ATOMIC_OP_ADD,
+                                                  sbuf, 1, target, rkey, &param);
+        ASSERT_FALSE(UCS_PTR_IS_ERR(sptr));
+        ASSERT_NE(sender().ep()->ext->unflushed_lanes, 0);
+        request_release(sptr);
+    }
+
+    void test_ep_based_fence(uint64_t size) {
+        mem_buffer sbuf(size, UCS_MEMORY_TYPE_HOST);
+        mapped_buffer rbuf(size, receiver());
+        bool is_fence_required;
+        uint64_t iter;
+
+        rbuf.memset(0);
+        sbuf.memset(CHAR_MAX);
+
+        ucs::handle<ucp_rkey_h> rkey;
+        rbuf.rkey(sender(), rkey);
+
+        for (iter = 1; iter < size; iter *= 10) {
+            put_nbx(sbuf.ptr(), sbuf.size(), (uint64_t)rbuf.ptr(), rkey);
+            do_fence();
+            is_fence_required = ucp_ep_is_fence_required(sender().ep()) &&
+                                ucp_ep_is_strong_fence(sender().ep());
+            put_nbx(sbuf.ptr(), sbuf.size(), (uint64_t)rbuf.ptr(), rkey);
+
+            if (is_fence_required) {
+                ASSERT_EQ(sender().ep()->worker->fence_seq,
+                          sender().ep()->ext->fence_seq);
+            } else {
+                ASSERT_NE(sender().ep()->worker->fence_seq,
+                          sender().ep()->ext->fence_seq);
+            }
+        }
+
+        for (iter = 1; iter < size; iter *= 10) {
+            get_nbx(sbuf.ptr(), sbuf.size(), (uint64_t)rbuf.ptr(), rkey);
+            do_fence();
+            is_fence_required = ucp_ep_is_fence_required(sender().ep()) &&
+                                ucp_ep_is_strong_fence(sender().ep());
+            get_nbx(sbuf.ptr(), sbuf.size(), (uint64_t)rbuf.ptr(), rkey);
+
+            if (is_fence_required) {
+                ASSERT_EQ(sender().ep()->worker->fence_seq,
+                          sender().ep()->ext->fence_seq);
+            } else {
+                ASSERT_NE(sender().ep()->worker->fence_seq,
+                          sender().ep()->ext->fence_seq);
+            }
+        }
+
+        atomic_op_nbx(sbuf.ptr(), sizeof(uint64_t), (uint64_t)rbuf.ptr(), rkey);
+        do_fence();
+        is_fence_required = ucp_ep_is_fence_required(sender().ep()) &&
+                            ucp_ep_is_strong_fence(sender().ep());
+        atomic_op_nbx(sbuf.ptr(), sizeof(uint64_t), (uint64_t)rbuf.ptr(), rkey);
+
+        if (is_fence_required) {
+            ASSERT_EQ(sender().ep()->worker->fence_seq,
+                      sender().ep()->ext->fence_seq);
+        } else {
+            ASSERT_NE(sender().ep()->worker->fence_seq,
+                      sender().ep()->ext->fence_seq);
+        }
+
+        flush_workers();
+    }
+};
+
+UCS_TEST_P(test_ucp_ep_based_fence, ep_based_fence) {
+    test_ep_based_fence(1000000);
+}
+
+UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_ep_based_fence, all, "all")