Skip to content

UCP/CORE: Implement flush+destroy for UCT EPs on UCP Worker #5608

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 34 commits into from
Sep 10, 2020
Merged

UCP/CORE: Implement flush+destroy for UCT EPs on UCP Worker #5608

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
34 commits
Select commit Hold shift + click to select a range
f7b0e3c
UCP/CORE: Implement flush+destroy for UCT EPs on UCP Worker
dmitrygx Aug 24, 2020
627a8a5
UCP/GTEST: Complete worker_flus operation if there is no flush ops an…
dmitrygx Aug 26, 2020
e9d3b1d
UCT/IB: Fix uct_iface_flush()/uct_ep_flush(LOCAL) after uct_ep_flush(…
dmitrygx Aug 26, 2020
ecd8663
UCT/IB/UD: Reset max_psn instead of stopping TX
dmitrygx Aug 26, 2020
8d4a9f9
UCM/GTEST: Fix leftovers
dmitrygx Aug 27, 2020
d229eae
UCP/WORKER: Introduce khash to find whether UCT EP there or not
dmitrygx Aug 27, 2020
8558d4c
UCP/CORE: Fix Coverity issue
dmitrygx Aug 28, 2020
cdfe88d
UCP/UCT: Fix review comments
dmitrygx Aug 28, 2020
c1a8fee
UCP/CORE: Fix review comments
dmitrygx Aug 31, 2020
28edf67
UCP/CORE: Fix EP-by-EP flush when iface_flush returns NO_RESOURCE
dmitrygx Aug 31, 2020
022a6f2
UCP/RMA/FLUSH: Fix flush ops count check usage
dmitrygx Sep 1, 2020
46b1272
UCP/CORE: Fix tests
dmitrygx Sep 2, 2020
cb197a4
UCP/CORE: Implemented purge
dmitrygx Sep 2, 2020
5d0b3fa
UCP/CORE: Fix bug in purging
dmitrygx Sep 3, 2020
12d1c34
UCP/CORE/GTEST: Fix review comments
dmitrygx Sep 3, 2020
d1ab7d4
GTEST/UCP: Fix review comments
dmitrygx Sep 3, 2020
d809dc1
UCP/CORE/GTEST: Fix bug in purging
dmitrygx Sep 3, 2020
7e6a5b4
Merge remote-tracking branch 'origin/master' into topic/ucp/worker_di…
dmitrygx Sep 3, 2020
a0f759b
UCP/CORE: Fix review comments
dmitrygx Sep 3, 2020
46afaaf
UCP/CORE/WIREUP/GTEST: Fix leak of WIREUP MSG proxy req
dmitrygx Sep 3, 2020
e2982d5
UCP/WIREUP/GTEST: Use pointer to the UCP request to be able free it i…
dmitrygx Sep 4, 2020
7ecd09d
GTEST/UCP: Fix review comments
dmitrygx Sep 4, 2020
9ef1f34
UCP/CORE: Fix review comments
dmitrygx Sep 4, 2020
efdb4d3
GTEST/UCP: Ensure that fluah+pending_add is registered on Worker prog…
dmitrygx Sep 6, 2020
509dd6c
UCP/GTEST: Fix review comments
dmitrygx Sep 7, 2020
07b8bb3
UCP/RMA: Remove flush_ops_count_check
dmitrygx Sep 7, 2020
43c3e92
UCP/GTEST: Fix review comments
dmitrygx Sep 8, 2020
9153e50
UCP/CORE: Added useful comments
dmitrygx Sep 8, 2020
9c346d6
UCP/RMA: Add more comments for flush to make it clear + fix typo
dmitrygx Sep 8, 2020
465be85
GTEST/UCP: Fix review comments
dmitrygx Sep 8, 2020
cdb8875
UCP/CORE: Use progress instead of loop over ERR_BUSY
dmitrygx Sep 8, 2020
2a5888f
UCP/GTEST: Fix review comments
dmitrygx Sep 9, 2020
d3cf051
UCP/WORKER: Fix review comments
dmitrygx Sep 9, 2020
ea18b51
UCP/CORE: Let UCP Wireup EP be destroyed in case of error
dmitrygx Sep 9, 2020
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
54 changes: 22 additions & 32 deletions src/ucp/core/ucp_worker.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2440,33 +2440,25 @@ ucp_worker_discard_uct_ep_flush_comp(uct_completion_t *self,
static ucs_status_t
ucp_worker_discard_uct_ep_pending_cb(uct_pending_req_t *self)
{
uct_worker_cb_id_t cb_id = UCS_CALLBACKQ_ID_NULL;
ucp_request_t *req = ucs_container_of(self, ucp_request_t, send.uct);
uct_ep_h uct_ep = req->send.discard_uct_ep.uct_ep;
ucp_worker_h worker = req->send.discard_uct_ep.ucp_worker;
ucs_status_t status, status_add;

status = uct_ep_flush(uct_ep, req->send.discard_uct_ep.ep_flush_flags,
&req->send.state.uct_comp);
if (status == UCS_ERR_NO_RESOURCE) {
status_add = uct_ep_pending_add(uct_ep, &req->send.uct, 0);
ucs_assert((status_add == UCS_ERR_BUSY) || (status_add == UCS_OK));
/* if added to the pending queue or to worker progress, need to remove
* the callback to not invoke it several times */
if (status_add == UCS_ERR_BUSY) {
uct_worker_progress_register_safe(worker->uct,
ucp_worker_discard_uct_ep_progress,
req, UCS_CALLBACKQ_FLAG_ONESHOT,
&cb_id);
}
ucp_request_t *req = ucs_container_of(self, ucp_request_t, send.uct);
uct_ep_h uct_ep = req->send.discard_uct_ep.uct_ep;
ucs_status_t status;

return UCS_OK;
} else if (status == UCS_INPROGRESS) {
/* need to remove from the pending queue */
status = UCS_OK;
} else {
ucp_worker_discard_uct_ep_flush_comp(&req->send.state.uct_comp, status);
}
do {
status = uct_ep_flush(uct_ep, req->send.discard_uct_ep.ep_flush_flags,
&req->send.state.uct_comp);
if (status == UCS_ERR_NO_RESOURCE) {
status = uct_ep_pending_add(uct_ep, &req->send.uct, 0);
ucs_assert((status == UCS_OK) || (status == UCS_ERR_BUSY));
} else if (status == UCS_INPROGRESS) {
/* need to remove from the pending queue */
status = UCS_OK;
} else {

@evgeny-leksikov evgeny-leksikov Sep 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

UCS_OK is not handled?

@dmitrygx dmitrygx Sep 8, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

it is handled in the else section

ucs_assert(status != UCS_ERR_BUSY);

@evgeny-leksikov evgeny-leksikov Sep 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

this assert looks wrong since status value is returned by flush here

@dmitrygx dmitrygx Sep 8, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

this is not wrong, just to make sure that we will not enter the endless loop if uct_ep_flush() returns UCS_ERR_BUSY
added the comment

@evgeny-leksikov evgeny-leksikov Sep 8, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

but uct_ep_flush doc does not say that it cannot return UCS_ERR_BUSY

@dmitrygx dmitrygx Sep 8, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

but uct_ep_flush doc does not say that it cannot return UCS_ERR_BUSY

yes, will handle it

@yosefe what do you think to use progress_register() instead of this loop?
seems to be error-prone or overcomplicated if it will have several retrun/break/continue in the loop

ucp_worker_discard_uct_ep_flush_comp(&req->send.state.uct_comp,
status);
}
Comment thread
yosefe marked this conversation as resolved.
} while (status == UCS_ERR_BUSY);

return status;

@yosefe yosefe Sep 9, 2020

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.

if status != NO_RESOURCE should also return UCS_OK

@yosefe yosefe Sep 9, 2020

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.

if flush returned ok, who will call the completion?

@dmitrygx dmitrygx Sep 9, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

if status != NO_RESOURCE should also return UCS_OK

why? we should put to the pending and wait for the callback invocation

if flush returned ok, who will call the completion?

it will be done in ucp_worker_discard_uct_ep_flush_comp()

@yosefe yosefe Sep 9, 2020

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.

  • if the status is error which is not NO_RESOURCE, the flush failed, we cannot retry
  • ok

@dmitrygx dmitrygx Sep 9, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

  • if the status is error which is not NO_RESOURCE, the flush failed, we cannot retry

ah, I see. fixed

}
Expand All @@ -2492,9 +2484,7 @@ ucp_worker_discard_wireup_ep(ucp_worker_h worker,
ucs_assert(wireup_ep != NULL);
ucs_assert(purge_cb != NULL);

ucp_wireup_ep_pending_purge_common(wireup_ep,
purge_cb, purge_arg,
purge_cb, purge_arg);
uct_ep_pending_purge(&wireup_ep->super.super, purge_cb, purge_arg);

if (wireup_ep->aux_ep != NULL) {
/* make sure that there is no WIREUP MSGs anymore */
Expand All @@ -2513,9 +2503,9 @@ ucp_worker_discard_wireup_ep(ucp_worker_h worker,
is_owner = wireup_ep->super.is_owner;
uct_ep = ucp_wireup_ep_extract_next_ep(&wireup_ep->super.super);

/* destroy WIREUP EP allocated for this UCT EP, since
* discard operation most likely won't have an access to
* UCP EP as it could be destroyed by the caller */
/* destroy WIREUP EP allocated for this UCT EP, since discard operation
* most likely won't have an access to UCP EP as it could be destroyed
* by the caller */
uct_ep_destroy(&wireup_ep->super.super);

/* do nothing, if this wireup EP is not an owner for UCT EP */
Expand Down
90 changes: 13 additions & 77 deletions src/ucp/wireup/wireup_ep.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,18 +22,6 @@
#include <ucp/core/ucp_request.inl>


typedef struct ucp_wireup_ep_pending_purge_arg {
ucp_wireup_ep_t *wireup_ep; /* WIREUP EP that owns the UCT
* pending request for the WIREUP
* MSG */
uct_pending_purge_callback_t wireup_msg_cb; /* UCT pending purge cakkback that
* will handle WIREUP MSG pending
* request */
void *wireup_msg_arg; /* UCT pending purge callback
* argument */
} ucp_wireup_ep_pending_purge_arg_t;


UCS_CLASS_DECLARE(ucp_wireup_ep_t, ucp_ep_h);


Expand Down Expand Up @@ -160,11 +148,7 @@ ucs_status_t ucp_wireup_ep_progress_pending(uct_pending_req_t *self)

status = req->func(req);
if (status == UCS_OK) {
/* WIREUP EP pointer could be NULL in case of it was discarded, but this
* pending request was completed on the UCT EP or the WIREUP AUX EP */
if (wireup_ep != NULL) {
ucs_atomic_sub32(&wireup_ep->pending_count, 1);
}
ucs_atomic_sub32(&wireup_ep->pending_count, 1);
ucs_free(proxy_req);
}
return status;
Expand Down Expand Up @@ -235,78 +219,30 @@ static ucs_status_t ucp_wireup_ep_pending_add(uct_ep_h uct_ep,
}

static void
ucp_wireup_ep_pending_purge_cb(uct_pending_req_t *self, void *arg)
{
ucp_wireup_ep_pending_purge_arg_t *purge_arg =
(ucp_wireup_ep_pending_purge_arg_t*)arg;
ucp_wireup_ep_t *wireup_ep = purge_arg->wireup_ep;
uct_ep_h uct_ep =
ucp_wireup_ep_get_msg_ep(wireup_ep);
ucp_request_t *wireup_proxy_req =
ucs_container_of(self, ucp_request_t, send.uct);

/* do purging on AUX EP or on UCT EP if WIREUP is an owner of it */
if ((uct_ep == wireup_ep->aux_ep) || wireup_ep->super.is_owner) {
/* need to NULL the WIREUP EP in the WIREUP MSG proxy pending request
* to avoid dereferencing it when progressing prnding requests, since
* it will be destroyed */
wireup_proxy_req->send.proxy.wireup_ep = NULL;

purge_arg->wireup_msg_cb(self, purge_arg->wireup_msg_arg);
if (purge_arg->wireup_msg_cb != ucp_wireup_ep_pending_req_release) {
/* decrement the pending count, if it is not a request release
* callback */
ucs_atomic_sub32(&wireup_ep->pending_count, 1);
}
}
}

void
ucp_wireup_ep_pending_purge_common(ucp_wireup_ep_t *wireup_ep,
uct_pending_purge_callback_t wireup_msg_cb,
void *wireup_msg_arg,
uct_pending_purge_callback_t user_msg_cb,
void *user_msg_arg)
ucp_wireup_ep_pending_purge(uct_ep_h uct_ep, uct_pending_purge_callback_t cb,
void *arg)
{
ucp_wireup_ep_pending_purge_arg_t purge_arg;
ucp_worker_h worker;
ucp_wireup_ep_t *wireup_ep = ucp_wireup_ep(uct_ep);
ucp_worker_h worker;
uct_pending_req_t *req;
ucp_request_t *ucp_req;
uct_ep_h uct_ep;
ucp_request_t *ucp_req;

worker = wireup_ep->super.ucp_ep->worker;

if (wireup_ep->pending_count > 0) {
uct_ep = ucp_wireup_ep_get_msg_ep(wireup_ep);

purge_arg.wireup_ep = wireup_ep;
purge_arg.wireup_msg_cb = wireup_msg_cb;
purge_arg.wireup_msg_arg = wireup_msg_arg;

uct_ep_pending_purge(uct_ep,
ucp_wireup_ep_pending_purge_cb,
&purge_arg);
}

ucs_assert(wireup_ep->pending_count == 0);

ucs_queue_for_each_extract(req, &wireup_ep->pending_q, priv, 1) {
ucp_req = ucs_container_of(req, ucp_request_t, send.uct);
UCS_ASYNC_BLOCK(&worker->async);
--worker->flush_ops_count;
UCS_ASYNC_UNBLOCK(&worker->async);
user_msg_cb(&ucp_req->send.uct, user_msg_arg);
cb(&ucp_req->send.uct, arg);
}
}

static void
ucp_wireup_ep_pending_purge(uct_ep_h uct_ep, uct_pending_purge_callback_t cb,
void *arg)
{
ucp_wireup_ep_t *wireup_ep = ucp_wireup_ep(uct_ep);
ucp_wireup_ep_pending_purge_common(wireup_ep,
ucp_wireup_ep_pending_req_release, NULL,
cb, arg);
if (wireup_ep->pending_count > 0) {
uct_ep_pending_purge(ucp_wireup_ep_get_msg_ep(wireup_ep),
ucp_wireup_ep_pending_req_release, arg);
}

ucs_assert(wireup_ep->pending_count == 0);
}

static ssize_t ucp_wireup_ep_am_bcopy(uct_ep_h uct_ep, uint8_t id,
Expand Down
7 changes: 0 additions & 7 deletions src/ucp/wireup/wireup_ep.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,13 +104,6 @@ void ucp_wireup_ep_disown(uct_ep_h uct_ep, uct_ep_h owned_ep);

ucs_status_t ucp_wireup_ep_progress_pending(uct_pending_req_t *self);

void
ucp_wireup_ep_pending_purge_common(ucp_wireup_ep_t *wireup_ep,
uct_pending_purge_callback_t wireup_msg_cb,
void *wireup_msg_arg,
uct_pending_purge_callback_t user_msg_cb,
void *user_msg_arg);

void ucp_wireup_ep_replay_pending_requests(ucp_ep_h ucp_ep,
ucs_queue_head_t *tmp_pending_queue);

Expand Down
70 changes: 15 additions & 55 deletions test/gtest/ucp/test_ucp_worker.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,16 +55,10 @@ class test_ucp_worker_discard : public ucp_test {
"ucp_request"));
ASSERT_TRUE(req != NULL);

pending_reqs[base + i] = req;
pending_reqs.push_back(req);

if (func == ucp_wireup_msg_progress) {

req->send.ep = &m_fake_ep;

/* for fast completing the WIREUP MSG, it frees the send
* buffer and returns UCS_OK */
req->send.wireup.type = UCP_WIREUP_MSG_REQUEST;
ASSERT_TRUE(m_fake_ep.flags & UCP_EP_FLAG_REMOTE_CONNECTED);
if (func == ucp_wireup_msg_progress) {
req->send.ep = &m_fake_ep;
}

req->send.uct.func = func;
Expand Down Expand Up @@ -104,25 +98,7 @@ class test_ucp_worker_discard : public ucp_test {
eps[i].iface = &iface;
m_created_ep_count++;

unsigned expected_pending_purge_reqs_count = 0;
unsigned added_pending_purge_reqs_count = 0;

if (ep_pending_purge_func == ep_pending_purge_func_iter_reqs) {
/* expected purging count is the number of pending
* requests in the WIREUP EP (if used) and in the
* WIREUP AUX EP (if used) or UCT EP (if no WIREUP EP
* or no WIREUP AUX EP) */
expected_pending_purge_reqs_count +=
m_pending_purge_reqs_count;

if (i < wireup_ep_count) {
expected_pending_purge_reqs_count +=
m_pending_purge_reqs_count;
}
}

std::vector<ucp_request_t*>
pending_reqs(expected_pending_purge_reqs_count);
std::vector<ucp_request_t*> pending_reqs;

if (i < wireup_ep_count) {
status = ucp_wireup_ep_create(&ucp_ep, &discard_ep);
Expand All @@ -144,15 +120,13 @@ class test_ucp_worker_discard : public ucp_test {
m_created_ep_count++;
}

if (expected_pending_purge_reqs_count > 0) {
if (ep_pending_purge_func == ep_pending_purge_func_iter_reqs) {
/* add WIREUP MSGs to the WIREUP EP (it will be added to
* UCT EP or WIREUP AUX EP) */
add_pending_reqs(discard_ep,
(uct_pending_callback_t)
ucp_wireup_msg_progress,
pending_reqs);
added_pending_purge_reqs_count +=
m_pending_purge_reqs_count;
}
} else {
discard_ep = &eps[i];
Expand All @@ -161,26 +135,25 @@ class test_ucp_worker_discard : public ucp_test {
EXPECT_LE(m_created_ep_count, total_ep_count);


if (expected_pending_purge_reqs_count > 0) {
if (ep_pending_purge_func == ep_pending_purge_func_iter_reqs) {
/* add user's pending requests */
add_pending_reqs(discard_ep,
(uct_pending_callback_t)
ucs_empty_function,
pending_reqs,
added_pending_purge_reqs_count);
added_pending_purge_reqs_count +=
m_pending_purge_reqs_count;
pending_reqs);
}

EXPECT_EQ(expected_pending_purge_reqs_count,
added_pending_purge_reqs_count);

unsigned purged_reqs_count = 0;
ucp_worker_discard_uct_ep(sender().worker(), discard_ep,
UCT_FLUSH_FLAG_LOCAL,
ep_pending_purge_count_reqs_cb,
&purged_reqs_count);
EXPECT_EQ(expected_pending_purge_reqs_count, purged_reqs_count);

if (ep_pending_purge_func == ep_pending_purge_func_iter_reqs) {
EXPECT_EQ(m_pending_purge_reqs_count, purged_reqs_count);
} else {
EXPECT_EQ(0u, purged_reqs_count);
}
}

void *flush_req = sender().flush_worker_nb(0);
Expand Down Expand Up @@ -319,21 +292,8 @@ class test_ucp_worker_discard : public ucp_test {
ucp_request_t,
send.uct);

if (self->func == ucp_wireup_ep_progress_pending) {
/* need to complete WIREUP MSG to release allocated
* proxy request */
/* TODO: replace by `ucp_request_send()` when
* `ucp/core/ucp_request.inl` file could be compiled
* by C++ compiler */
ucs_status_t status = req->send.uct.func(&req->send.uct);
EXPECT_EQ(UCS_OK, status);
/* no need to release the memory allocated for the request.
* it will be freed in the `ucp_wireup_msg_send()` function,
* since it expects that the request was allocated in the
* `ucp_wireup_msg_send()` function */
} else {
ucs_free(req);
}
ASSERT_TRUE(self->func != ucp_wireup_ep_progress_pending);
ucs_free(req);
}

static ucs_status_t
Expand Down