Skip to content
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

Raise UCXNotConnected on stream receive callback #799

Merged
merged 1 commit into from
Oct 21, 2021
Merged

Raise UCXNotConnected on stream receive callback #799

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
12 changes: 11 additions & 1 deletion ucp/_libs/transfer_stream.pyx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,13 @@ from libc.stdint cimport uintptr_t
from .arr cimport Array
from .ucx_api_dep cimport *

from ..exceptions import UCXCanceled, UCXError, UCXMsgTruncated, log_errors
from ..exceptions import (
UCXCanceled,
UCXError,
UCXMsgTruncated,
UCXNotConnected,
log_errors,
)


def stream_send_nb(
Expand Down Expand Up @@ -113,6 +119,10 @@ cdef void _stream_recv_callback(
name = req_info["name"]
msg = "<%s>: " % name
exception = UCXCanceled(msg)
if status == UCS_ERR_NOT_CONNECTED:
Copy link
Member

Choose a reason for hiding this comment

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

Should we add any debug logging here ? If it's intermittent maybe this might help

Copy link
Member Author

Choose a reason for hiding this comment

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

I mean the Dask use behavior is intermittent, not the code itself. The problem seems to be that when Dask tries to reconnect this error is raised, but the reconnect process happens once the other end has already closed. The exception can be seen clearly when it happens, and this is what dask/distributed#5449 is catching now, therefore logging doesn't seem to have any use to me now.

name = req_info["name"]
msg = "<%s>: " % name
exception = UCXNotConnected(msg)
elif status != UCS_OK:
name = req_info["name"]
ucx_status_msg = ucs_status_string(status).decode("utf-8")
Expand Down
1 change: 1 addition & 0 deletions ucp/_libs/ucx_api_dep.pxd
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,7 @@ cdef extern from "ucp/api/ucp.h":
ucs_status_t UCS_ERR_NO_ELEM
ucs_status_t UCS_ERR_BUSY
ucs_status_t UCS_ERR_CANCELED
ucs_status_t UCS_ERR_NOT_CONNECTED
ucs_status_t UCS_ERR_CONNECTION_RESET

void ucp_get_version(unsigned * major_version,
Expand Down
4 changes: 4 additions & 0 deletions ucp/exceptions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,3 +47,7 @@ class UCXConnectionReset(UCXBaseException):

class UCXMsgTruncated(UCXBaseException):
pass


class UCXNotConnected(UCXBaseException):
pass