diff --git a/src/ucp/core/ucp_request.c b/src/ucp/core/ucp_request.c index cfe0d573fca..d9bd8cdf760 100644 --- a/src/ucp/core/ucp_request.c +++ b/src/ucp/core/ucp_request.c @@ -600,6 +600,9 @@ void ucp_request_send_state_ff(ucp_request_t *req, ucs_status_t status) req->send.proto.comp_cb(req); } else if (req->send.state.uct_comp.func == ucp_ep_flush_completion) { ucp_ep_flush_request_ff(req, status); + } else if (req->send.state.uct_comp.func == + ucp_worker_discard_uct_ep_flush_comp) { + ucp_worker_discard_uct_ep_progress(req); } else if (req->send.state.uct_comp.func != NULL) { /* Fast-forward the sending state to complete the operation when last * network completion callback is called diff --git a/src/ucp/core/ucp_worker.c b/src/ucp/core/ucp_worker.c index b2d28dc6771..eca3335fc64 100644 --- a/src/ucp/core/ucp_worker.c +++ b/src/ucp/core/ucp_worker.c @@ -581,6 +581,13 @@ ucp_worker_iface_error_handler(void *arg, uct_ep_h uct_ep, ucs_status_t status) if (ucp_worker_is_uct_ep_discarding(worker, uct_ep)) { ucs_debug("UCT EP %p is being discarded on UCP Worker %p", uct_ep, worker); + /* FLUSH_CANCEL operation might be on pending queue due to + * UCS_ERR_NO_RESOURCES, so need to purge the queue to resubmit the + * operation. We need to resubmit the FLUSH_CANCEL operation on the same + * failed lane, in order to make sure all previous outstanding + * operations are completed before destroying the failed endpoint. */ + uct_ep_pending_purge(uct_ep, ucp_ep_err_pending_purge, + UCS_STATUS_PTR(UCS_ERR_CANCELED)); ret_status = UCS_OK; goto out; } @@ -2298,7 +2305,7 @@ static void ucp_worker_discard_uct_ep_progress_register(ucp_request_t *req, &req->send.discard_uct_ep.cb_id); } -static void ucp_worker_discard_uct_ep_flush_comp(uct_completion_t *self) +void ucp_worker_discard_uct_ep_flush_comp(uct_completion_t *self) { ucp_request_t *req = ucs_container_of(self, ucp_request_t, send.state.uct_comp); @@ -2322,12 +2329,11 @@ ucp_worker_discard_uct_ep_pending_cb(uct_pending_req_t *self) status = uct_ep_flush(uct_ep, req->send.discard_uct_ep.ep_flush_flags, &req->send.state.uct_comp); if (status == UCS_OK) { - /* don't destroy UCT EP from the pending callback, schedule a progress - * callback on the main thread to destroy UCT EP */ - ucp_worker_discard_uct_ep_progress_register( - req, ucp_worker_discard_uct_ep_destroy_progress); + ucs_assert(req->send.state.uct_comp.count == 0); + ucp_worker_discard_uct_ep_flush_comp(&req->send.state.uct_comp); return UCS_OK; } else if (status == UCS_INPROGRESS) { + req->send.state.uct_comp.count++; return UCS_OK; } else if (status == UCS_ERR_NO_RESOURCE) { return UCS_ERR_NO_RESOURCE; @@ -2338,7 +2344,7 @@ ucp_worker_discard_uct_ep_pending_cb(uct_pending_req_t *self) return UCS_OK; } -static unsigned ucp_worker_discard_uct_ep_progress(void *arg) +unsigned ucp_worker_discard_uct_ep_progress(void *arg) { ucp_request_t *req = (ucp_request_t*)arg; uct_ep_h uct_ep = req->send.discard_uct_ep.uct_ep; @@ -3038,7 +3044,7 @@ static void ucp_worker_discard_tl_uct_ep(ucp_ep_h ucp_ep, uct_ep_h uct_ep, req->send.ep = ucp_ep; req->send.uct.func = ucp_worker_discard_uct_ep_pending_cb; req->send.state.uct_comp.func = ucp_worker_discard_uct_ep_flush_comp; - req->send.state.uct_comp.count = 1; + req->send.state.uct_comp.count = 0; req->send.state.uct_comp.status = UCS_OK; req->send.discard_uct_ep.uct_ep = uct_ep; req->send.discard_uct_ep.ep_flush_flags = ep_flush_flags; diff --git a/src/ucp/core/ucp_worker.h b/src/ucp/core/ucp_worker.h index d7537a56a3d..907e6ac677f 100644 --- a/src/ucp/core/ucp_worker.h +++ b/src/ucp/core/ucp_worker.h @@ -365,6 +365,12 @@ char *ucp_worker_print_used_tls(const ucp_ep_config_key_t *key, ucp_worker_cfg_index_t config_idx, char *info, size_t max); +void ucp_worker_vfs_refresh(void *obj); + +void ucp_worker_discard_uct_ep_flush_comp(uct_completion_t *self); + +unsigned ucp_worker_discard_uct_ep_progress(void *arg); + static UCS_F_ALWAYS_INLINE void ucp_worker_flush_ops_count_inc(ucp_worker_h worker) { @@ -382,6 +388,4 @@ ucp_worker_flush_ops_count_dec(ucp_worker_h worker) --worker->flush_ops_count; } -void ucp_worker_vfs_refresh(void *obj); - #endif diff --git a/test/gtest/ucp/test_ucp_peer_failure.cc b/test/gtest/ucp/test_ucp_peer_failure.cc index 7155e147d8e..c263c29d477 100644 --- a/test/gtest/ucp/test_ucp_peer_failure.cc +++ b/test/gtest/ucp/test_ucp_peer_failure.cc @@ -43,7 +43,6 @@ class test_ucp_peer_failure : public ucp_test { static ucs_status_t am_callback(void *arg, const void *header, size_t header_length, void *data, size_t length, const ucp_am_recv_param_t *param); - void set_timeouts(); static void err_cb(void *arg, ucp_ep_h ep, ucs_status_t status); ucp_ep_h stable_sender(); ucp_ep_h failing_sender(); @@ -75,7 +74,6 @@ class test_ucp_peer_failure : public ucp_test { std::string m_sbuf, m_rbuf; mem_handle_t m_stable_memh, m_failing_memh; ucs::handle m_stable_rkey, m_failing_rkey; - ucs::ptr_vector m_env; }; UCP_INSTANTIATE_TEST_CASE(test_ucp_peer_failure) @@ -85,7 +83,7 @@ test_ucp_peer_failure::test_ucp_peer_failure() : m_am_rx_count(0), m_err_count(0), m_err_status(UCS_OK) { ucs::fill_random(m_sbuf); - set_timeouts(); + set_tl_small_timeouts(); } void test_ucp_peer_failure::get_test_variants( @@ -139,10 +137,6 @@ test_ucp_peer_failure::am_callback(void *arg, const void *header, return UCS_OK; } -void test_ucp_peer_failure::set_timeouts() { - set_tl_timeouts(m_env); -} - void test_ucp_peer_failure::err_cb(void *arg, ucp_ep_h ep, ucs_status_t status) { test_ucp_peer_failure *self = reinterpret_cast(arg); EXPECT_TRUE((UCS_ERR_CONNECTION_RESET == status) || diff --git a/test/gtest/ucp/test_ucp_sockaddr.cc b/test/gtest/ucp/test_ucp_sockaddr.cc index ffaa69e1c4a..5ba0dac6f83 100644 --- a/test/gtest/ucp/test_ucp_sockaddr.cc +++ b/test/gtest/ucp/test_ucp_sockaddr.cc @@ -67,14 +67,6 @@ class test_ucp_sockaddr : public ucp_test { modify_config("KEEPALIVE_INTERVAL", "10s"); modify_config("CM_USE_ALL_DEVICES", cm_use_all_devices() ? "y" : "n"); - /* - * FIXME: this is a workaround of the issue reproduced by - * the 'close_ep_force_before_err_cb' test with RC transport - * where TX-queue less than FC_WND, so uct_flush cancel returns - * UCS_ERR_NO_RESOURCES and can not be handled after error - */ - modify_config("RC_FC_WND_SIZE", "128", SETENV_IF_NOT_EXIST); - get_sockaddr(); ucp_test::init(); skip_loopback(); @@ -1351,18 +1343,19 @@ UCS_TEST_P(test_ucp_sockaddr_check_lanes, check_rndv_lanes) UCP_INSTANTIATE_ALL_TEST_CASE(test_ucp_sockaddr_check_lanes) + class test_ucp_sockaddr_destroy_ep_on_err : public test_ucp_sockaddr { public: test_ucp_sockaddr_destroy_ep_on_err() { - set_tl_timeouts(m_env); + set_tl_small_timeouts(); } virtual ucp_ep_params_t get_server_ep_params() { ucp_ep_params_t params = test_ucp_sockaddr::get_server_ep_params(); params.field_mask |= UCP_EP_PARAM_FIELD_ERR_HANDLING_MODE | - UCP_EP_PARAM_FIELD_ERR_HANDLER | - UCP_EP_PARAM_FIELD_USER_DATA; + UCP_EP_PARAM_FIELD_ERR_HANDLER | + UCP_EP_PARAM_FIELD_USER_DATA; params.err_mode = UCP_ERR_HANDLING_MODE_PEER; params.err_handler.cb = err_handler_cb; params.err_handler.arg = NULL; @@ -1375,9 +1368,6 @@ class test_ucp_sockaddr_destroy_ep_on_err : public test_ucp_sockaddr { entity *e = reinterpret_cast(arg); e->disconnect_nb(0, 0, UCP_EP_CLOSE_MODE_FORCE); } - -private: - ucs::ptr_vector m_env; }; UCS_TEST_P(test_ucp_sockaddr_destroy_ep_on_err, empty) { @@ -2205,7 +2195,7 @@ class test_ucp_sockaddr_protocols_err : public test_ucp_sockaddr_protocols { protected: test_ucp_sockaddr_protocols_err() { - set_tl_timeouts(m_env); + set_tl_small_timeouts(); } void test_tag_send_recv(size_t size, bool is_exp, @@ -2219,8 +2209,6 @@ class test_ucp_sockaddr_protocols_err : public test_ucp_sockaddr_protocols { variants & SEND_STOP, variants & RECV_STOP); } - - ucs::ptr_vector m_env; }; @@ -2302,7 +2290,7 @@ class test_ucp_sockaddr_protocols_err_sender } test_ucp_sockaddr_protocols_err_sender() { - set_tl_timeouts(m_env); + set_tl_small_timeouts(); m_env.push_back(new ucs::scoped_setenv("UCX_IB_REG_METHODS", "rcache,odp,direct")); } diff --git a/test/gtest/ucp/ucp_test.cc b/test/gtest/ucp/ucp_test.cc index 592bdcba9df..618c8a2bfaa 100644 --- a/test/gtest/ucp/ucp_test.cc +++ b/test/gtest/ucp/ucp_test.cc @@ -297,12 +297,12 @@ int ucp_test::max_connections() { } } -void ucp_test::set_tl_timeouts(ucs::ptr_vector &env) +void ucp_test::set_tl_small_timeouts() { /* Set small TL timeouts to reduce testing time */ - env.push_back(new ucs::scoped_setenv("UCX_RC_TIMEOUT", "10ms")); - env.push_back(new ucs::scoped_setenv("UCX_RC_RNR_TIMEOUT", "10ms")); - env.push_back(new ucs::scoped_setenv("UCX_RC_RETRY_COUNT", "2")); + m_env.push_back(new ucs::scoped_setenv("UCX_RC_TIMEOUT", "10ms")); + m_env.push_back(new ucs::scoped_setenv("UCX_RC_RNR_TIMEOUT", "10ms")); + m_env.push_back(new ucs::scoped_setenv("UCX_RC_RETRY_COUNT", "2")); } void ucp_test::set_ucp_config(ucp_config_t *config, const std::string& tls) diff --git a/test/gtest/ucp/ucp_test.h b/test/gtest/ucp/ucp_test.h index 5f3d79e07ba..08ded51970d 100644 --- a/test/gtest/ucp/ucp_test.h +++ b/test/gtest/ucp/ucp_test.h @@ -243,7 +243,7 @@ class ucp_test : public ucp_test_base, int worker_index = 0); void request_release(void *req); int max_connections(); - void set_tl_timeouts(ucs::ptr_vector &env); + void set_tl_small_timeouts(); // Add test variant without values, with given context params static ucp_test_variant& @@ -339,6 +339,8 @@ class ucp_test : public ucp_test_base, ucp_mem_h m_memh; void* m_rkey_buffer; }; + + ucs::ptr_vector m_env; };