Skip to content

UCP/EP: fix discarding from pending on failed lane #6933

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yosefe merged 2 commits into from
Jun 25, 2021
Merged

UCP/EP: fix discarding from pending on failed lane #6933

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
212a0c9
UCP/EP: fix discarding from pending on failed lane
evgeny-leksikov Jun 11, 2021
8bfb149
GTEST/UCP/SOCKADDR: remove workaround for FC window / TX-queue length
evgeny-leksikov Jun 10, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/ucp/core/ucp_request.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 13 additions & 7 deletions src/ucp/core/ucp_worker.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,

@dmitrygx dmitrygx Jun 22, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why needed?
UCT EP should be purged as a part of discarding procedure

@evgeny-leksikov evgeny-leksikov Jun 22, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

discard itself can be on pending

@dmitrygx dmitrygx Jun 22, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

discard itself can be on pending

but why we need to remove discarding from the pending?

@dmitrygx dmitrygx Jun 22, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it should be removed by discarding

@evgeny-leksikov evgeny-leksikov Jun 22, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it does not remove, ucp_ep_err_pending_purge does ucp_request_send_state_ff which posts flush cancel again, that's the fix

@yosefe yosefe Jun 23, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pls add comment to describe why it's needed

UCS_STATUS_PTR(UCS_ERR_CANCELED));
ret_status = UCS_OK;
goto out;
}
Expand Down Expand Up @@ -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);
Expand All @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Comment thread
yosefe marked this conversation as resolved.
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;
Expand Down
8 changes: 6 additions & 2 deletions src/ucp/core/ucp_worker.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
{
Expand All @@ -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
8 changes: 1 addition & 7 deletions test/gtest/ucp/test_ucp_peer_failure.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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<ucp_rkey_h> m_stable_rkey, m_failing_rkey;
ucs::ptr_vector<ucs::scoped_setenv> m_env;
};

UCP_INSTANTIATE_TEST_CASE(test_ucp_peer_failure)
Expand All @@ -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(
Expand Down Expand Up @@ -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<test_ucp_peer_failure*>(arg);
EXPECT_TRUE((UCS_ERR_CONNECTION_RESET == status) ||
Expand Down
24 changes: 6 additions & 18 deletions test/gtest/ucp/test_ucp_sockaddr.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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;
Expand All @@ -1375,9 +1368,6 @@ class test_ucp_sockaddr_destroy_ep_on_err : public test_ucp_sockaddr {
entity *e = reinterpret_cast<entity *>(arg);
e->disconnect_nb(0, 0, UCP_EP_CLOSE_MODE_FORCE);
}

private:
ucs::ptr_vector<ucs::scoped_setenv> m_env;
};

UCS_TEST_P(test_ucp_sockaddr_destroy_ep_on_err, empty) {
Expand Down Expand Up @@ -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,
Expand All @@ -2219,8 +2209,6 @@ class test_ucp_sockaddr_protocols_err : public test_ucp_sockaddr_protocols {
variants & SEND_STOP,
variants & RECV_STOP);
}

ucs::ptr_vector<ucs::scoped_setenv> m_env;
};


Expand Down Expand Up @@ -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"));
}
Expand Down
8 changes: 4 additions & 4 deletions test/gtest/ucp/ucp_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,12 +297,12 @@ int ucp_test::max_connections() {
}
}

void ucp_test::set_tl_timeouts(ucs::ptr_vector<ucs::scoped_setenv> &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)
Expand Down
4 changes: 3 additions & 1 deletion test/gtest/ucp/ucp_test.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<ucs::scoped_setenv> &env);
void set_tl_small_timeouts();

// Add test variant without values, with given context params
static ucp_test_variant&
Expand Down Expand Up @@ -339,6 +339,8 @@ class ucp_test : public ucp_test_base,
ucp_mem_h m_memh;
void* m_rkey_buffer;
};

ucs::ptr_vector<ucs::scoped_setenv> m_env;
};


Expand Down