Handle errors on individual workers in run/broadcast #5590

crusaderky · 2021-12-13T15:50:14Z

Add sensible handling of failures on individual workers:

Client.run

By default, continue raising as before. Add the option to return or ignore errors raised by the function on individual workers, so that the output from other workers is not lost. Deal with RPC errors in the same way as with exceptions raised by functions.

Scheduler.broadcast

In case of RPC errors, by default, continue raising as before. Add option to return or ignore the raised OSErrors instead, so that the output from other workers is not lost.

Client.dump_cluster_state

If a worker is unresponsive, dump the string repr of the RPC error alongside the _to_dict output of healthy workers

Client.version_info

If a worker is unresponsive, omit it and return the output of healthy workers

crusaderky · 2021-12-13T15:56:33Z

distributed/client.py

-            kwargs=dumps(kwargs, protocol=4),
+            function=dumps(function),
+            args=dumps(args),
+            kwargs=dumps(kwargs),


Bump to protocol 5

I think using protocol=4 here was to support cases where different nodes are running Python 3.7 (where protocol=5 isn't available) and 3.8+. To be clear, there may be other places where things break in this situation, but historically we've tried to be accommodating where possible (within reason). We'll probably drop Python 3.7 support soon -- it might be worth keeping protocol=4 for now and dropping it when we drop Python 3.7

crusaderky · 2021-12-13T16:01:41Z

distributed/client.py

@@ -3546,7 +3604,7 @@ def dump_cluster_state(
        filename: str = "dask-cluster-dump",
        exclude: Collection[str] = (),
        format: Literal["msgpack", "yaml"] = "msgpack",
-    ) -> Awaitable | None:
+    ):


The previous version would confuse mypy (see mypy guidelines re. return Any). I don't think there's an easy way to annotate sync/async client functions short of some major refactoring work.

jrbourbeau

Thanks @crusaderky. @ncclementi @scharlottej13 and I just took a look at this and overall it looks good, we left a couple of small comments below

jrbourbeau · 2021-12-13T21:13:16Z

distributed/client.py

-            kwargs=dumps(kwargs, protocol=4),
+            function=dumps(function),
+            args=dumps(args),
+            kwargs=dumps(kwargs),


I think using protocol=4 here was to support cases where different nodes are running Python 3.7 (where protocol=5 isn't available) and 3.8+. To be clear, there may be other places where things break in this situation, but historically we've tried to be accommodating where possible (within reason). We'll probably drop Python 3.7 support soon -- it might be worth keeping protocol=4 for now and dropping it when we drop Python 3.7

jrbourbeau · 2021-12-13T21:32:13Z

distributed/scheduler.py

        serializers=None,
-    ):
+        on_error: "Literal['raise', 'return', 'return_pickle', 'ignore']" = "raise",


Can we remove these quotes like in other places where we're using Literal?

Suggested change

on_error: "Literal['raise', 'return', 'return_pickle', 'ignore']" = "raise",

on_error: Literal['raise', 'return', 'return_pickle', 'ignore'] = "raise",

Not in this module, because Cython does not support from __future__ import annotations

Ah, I see -- thanks for clarifying. I should have known you would have done this is you could : )

jrbourbeau · 2021-12-13T22:05:00Z

distributed/scheduler.py

+            warnings.warn(
+                "workers=True is deprecated; pass workers=None or omit instead",
+                category=FutureWarning,
+            )


Could you add a small pytest.warns test for this?

jrbourbeau · 2021-12-13T22:23:22Z

Hrm, gpuCI is failing with

11:12:05 E   TypeError: broadcast() got an unexpected keyword argument 'on_error'

which is related because this PR is where on_error= is added to Scheduler.broadcast, but is odd because on_error is actually added, so I wouldn't expect this TypeError.

EDIT: We're getting a version mismatch warning for the failing test (see below). The scheduler in that test isn't running with the changes in this PR, which is why we're getting the unexpected keyword argument 'on_error' error. So while the changes here are causing test_ucx_config_w_env_var to fail, it's more so that the changes in this PR are revealing an issue which already existed in test_ucx_config_w_env_var. cc @charlesbluca @pentschev for visibility

11:12:05 =============================== warnings summary ===============================

11:12:05 distributed/comm/tests/test_ucx_config.py::test_ucx_config_w_env_var

11:12:05   /workspace/distributed/client.py:1096: VersionMismatchWarning: Mismatched versions found

11:12:05   

11:12:05   +-------------+------------------------+-----------+---------+

11:12:05   | Package     | client                 | scheduler | workers |

11:12:05   +-------------+------------------------+-----------+---------+

11:12:05   | distributed | 2021.12.0+11.g957be0fc | 2021.12.0 | None    |

11:12:05   +-------------+------------------------+-----------+---------+

11:12:05     warnings.warn(version_module.VersionMismatchWarning(msg[0]["warning"]))

crusaderky · 2021-12-13T22:25:58Z

Reverted to protocol=4. All review comments have been incorporated.

charlesbluca · 2021-12-13T22:43:45Z

Thanks for bringing this up @jrbourbeau! I unfortunately don't have much experience with this test and I believe @pentschev is offline; @quasiben does anything about these failures stand out to you?

jrbourbeau · 2021-12-13T22:44:13Z

distributed/client.py

@@ -3872,7 +3930,8 @@ async def _get_versions(self, check=False, packages=[]):
            scheduler = await self.scheduler.versions()  # this raises

        workers = await self.scheduler.broadcast(
-            msg={"op": "versions", "packages": packages}
+            msg={"op": "versions", "packages": packages},
+            on_error="ignore",


Sorry, one last comment after thinking about this a bit more. We often ask users to inspect the output of Client.get_versions when debugging package version mismatches. With on_error="ignore" here we could run into cases where there's a problematic version mismatch on an unresponsive worker which will be silently ignored (and hard to debug). What happens today? My guess is a connection error is raised

Thoughts on emitting a warning instead of silently ignoring?

Now logging an error in the scheduler log

pentschev · 2021-12-13T23:11:00Z

Hrm, gpuCI is failing with
11:12:05 E   TypeError: broadcast() got an unexpected keyword argument 'on_error'
which is related because this PR is where on_error= is added to Scheduler.broadcast, but is odd because on_error is actually added, so I wouldn't expect this TypeError.

EDIT: We're getting a version mismatch warning for the failing test (see below). The scheduler in that test isn't running with the changes in this PR, which is why we're getting the unexpected keyword argument 'on_error' error. So while the changes here are causing test_ucx_config_w_env_var to fail, it's more so that the changes in this PR are revealing an issue which already existed in test_ucx_config_w_env_var. cc @charlesbluca @pentschev for visibility
11:12:05 =============================== warnings summary ===============================

11:12:05 distributed/comm/tests/test_ucx_config.py::test_ucx_config_w_env_var

11:12:05   /workspace/distributed/client.py:1096: VersionMismatchWarning: Mismatched versions found

11:12:05   

11:12:05   +-------------+------------------------+-----------+---------+

11:12:05   | Package     | client                 | scheduler | workers |

11:12:05   +-------------+------------------------+-----------+---------+

11:12:05   | distributed | 2021.12.0+11.g957be0fc | 2021.12.0 | None    |

11:12:05   +-------------+------------------------+-----------+---------+

11:12:05     warnings.warn(version_module.VersionMismatchWarning(msg[0]["warning"]))

I can't reproduce the error locally neither with main nor this PR:

$ pytest -vs distributed/comm/tests/test_ucx_config.py
================================================== test session starts ==================================================
platform linux -- Python 3.8.12, pytest-6.2.5, py-1.11.0, pluggy-1.0.0 -- /datasets/pentschev/miniconda3/envs/rn-115-22.02.211213/bin/python
cachedir: .pytest_cache
rootdir: /datasets/pentschev/src/distributed, configfile: setup.cfg
plugins: asyncio-0.16.0, repeat-0.8.0, rerunfailures-10.2, timeout-2.0.1
timeout: 300.0s
timeout method: thread
timeout func_only: False
collected 2 items

distributed/comm/tests/test_ucx_config.py::test_ucx_config PASSED
distributed/comm/tests/test_ucx_config.py::test_ucx_config_w_env_var PASSED

================================================= slowest 20 durations ==================================================
2.65s call     distributed/comm/tests/test_ucx_config.py::test_ucx_config_w_env_var
0.01s call     distributed/comm/tests/test_ucx_config.py::test_ucx_config
0.01s setup    distributed/comm/tests/test_ucx_config.py::test_ucx_config
0.00s setup    distributed/comm/tests/test_ucx_config.py::test_ucx_config_w_env_var
0.00s teardown distributed/comm/tests/test_ucx_config.py::test_ucx_config_w_env_var
0.00s teardown distributed/comm/tests/test_ucx_config.py::test_ucx_config
=================================================== 2 passed in 3.07s ===================================================

$ ipython
Python 3.8.12 | packaged by conda-forge | (default, Oct 12 2021, 21:59:51)
Type 'copyright', 'credits' or 'license' for more information
IPython 7.30.1 -- An enhanced Interactive Python. Type '?' for help.

In [1]: import distributed

In [2]: distributed.__version__
Out[2]: '2021.12.0+13.g8a085126'

I don't quite get why there are mismatched versions here, but perhaps that's the issue then?

pentschev · 2021-12-13T23:12:41Z

Also as @charlesbluca mentioned, I'm indeed mostly off through the rest of the year, so my response may be severely delayed.

jrbourbeau · 2021-12-14T00:10:32Z

Thanks @charlesbluca @pentschev. #5595 should resolve the version warning

(Also, hope you enjoy your time off @pentschev 🙂)

crusaderky · 2021-12-14T10:05:19Z

All test failures (besides gpuCI) seem unrelated

…cast_error_handling

jrbourbeau · 2021-12-14T15:30:20Z

Just merged main which should fix the gpuCI error. @crusaderky thoughts on this comment #5590 (comment)?

crusaderky · 2021-12-16T10:54:58Z

All test failures are unrelated.
@jrbourbeau is there anything outstanding before this can be merged?

…cast_error_handling

crusaderky added 2 commits December 13, 2021 15:46

run/broadcast error handling

812d7d8

revert

385c500

crusaderky commented Dec 13, 2021

View reviewed changes

fix

8a6e9c2

crusaderky self-assigned this Dec 13, 2021

crusaderky mentioned this pull request Dec 13, 2021

Improved cluster state dump in @gen_cluster #5592

Merged

fix

8bef5e4

jrbourbeau reviewed Dec 13, 2021

View reviewed changes

Code review

35e2dba

Merge branch 'main' into broadcast_error_handling

8a08512

jrbourbeau reviewed Dec 13, 2021

View reviewed changes

jrbourbeau mentioned this pull request Dec 13, 2021

Updates to gpuCI and test_ucx_config_w_env_var #5595

Merged

Merge branch 'main' of https://github.com/dask/distributed into broad…

1e53dd5

…cast_error_handling

Code review

2c28466

Merge branch 'main' of https://github.com/dask/distributed into broad…

8ebecd1

…cast_error_handling

jrbourbeau approved these changes Dec 17, 2021

View reviewed changes

jrbourbeau merged commit ef5b088 into dask:main Dec 17, 2021

crusaderky deleted the broadcast_error_handling branch December 17, 2021 11:15

fjetter mentioned this pull request Jan 18, 2022

Deadlock after few workers take all tasks #5635

Open

jakirkham mentioned this pull request Feb 3, 2022

Drop support for cythonized scheduler #5685

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Handle errors on individual workers in run/broadcast #5590

Handle errors on individual workers in run/broadcast #5590

crusaderky commented Dec 13, 2021 •

edited

Loading

crusaderky Dec 13, 2021 •

edited

Loading

jrbourbeau Dec 13, 2021

crusaderky Dec 13, 2021 •

edited

Loading

jrbourbeau left a comment

jrbourbeau Dec 13, 2021

jrbourbeau Dec 13, 2021

crusaderky Dec 13, 2021 •

edited

Loading

jrbourbeau Dec 13, 2021

jrbourbeau Dec 13, 2021

crusaderky Dec 13, 2021

jrbourbeau commented Dec 13, 2021

crusaderky commented Dec 13, 2021

charlesbluca commented Dec 13, 2021

jrbourbeau Dec 13, 2021

crusaderky Dec 15, 2021

pentschev commented Dec 13, 2021 •

edited

Loading

pentschev commented Dec 13, 2021

jrbourbeau commented Dec 14, 2021

crusaderky commented Dec 14, 2021 •

edited

Loading

jrbourbeau commented Dec 14, 2021

crusaderky commented Dec 16, 2021

	on_error: "Literal['raise', 'return', 'return_pickle', 'ignore']" = "raise",
	on_error: Literal['raise', 'return', 'return_pickle', 'ignore'] = "raise",

Handle errors on individual workers in run/broadcast #5590

Handle errors on individual workers in run/broadcast #5590

Conversation

crusaderky commented Dec 13, 2021 • edited Loading

Client.run

Scheduler.broadcast

Client.dump_cluster_state

Client.version_info

crusaderky Dec 13, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

crusaderky Dec 13, 2021 • edited Loading

Choose a reason for hiding this comment

jrbourbeau left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

crusaderky Dec 13, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrbourbeau commented Dec 13, 2021

crusaderky commented Dec 13, 2021

charlesbluca commented Dec 13, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pentschev commented Dec 13, 2021 • edited Loading

pentschev commented Dec 13, 2021

jrbourbeau commented Dec 14, 2021

crusaderky commented Dec 14, 2021 • edited Loading

jrbourbeau commented Dec 14, 2021

crusaderky commented Dec 16, 2021

crusaderky commented Dec 13, 2021 •

edited

Loading

crusaderky Dec 13, 2021 •

edited

Loading

crusaderky Dec 13, 2021 •

edited

Loading

crusaderky Dec 13, 2021 •

edited

Loading

pentschev commented Dec 13, 2021 •

edited

Loading

crusaderky commented Dec 14, 2021 •

edited

Loading