Raise plugin exceptions on Worker.start() #4298

pentschev · 2020-12-02T18:46:24Z

This PR raises plugin exception in Worker.start() that were so far being silently ignored.

pentschev · 2020-12-02T18:49:06Z

distributed/worker.py

            *[self.plugin_add(plugin=plugin) for plugin in self._pending_plugins]
        )
+        for msg in plugins_msgs:
+            if msg["status"] != "OK":
+                raise msg["exception"].data


I'm not sure if this is the right way of raising the exception that's being created by distributed.core.error_message. The object in msg["exception"] is delivered as a distributed.protocol.Serialized, but there seems to be no way exposed to extract its data other than just addressing the data attributed directly.

I'm not sure if this is also being handled correctly or if this is a bug in error_message. Any other suggestions here are appreciated.

I suspect that if you just remove the try-except and special handling in the plugin_add function that things may just work. I suspect that the standard machinery around rpc will handle the exception comfortably.

I'm not sure though, this is just a guess. I give it a 50/50 chance.

I think your suggestion would work in Worker.start(), but plugin_add is used also as a handler in https://github.com/dask/distributed/blob/master/distributed/worker.py#L662 and I think that would break it. Of course, I could have two different functions, one that raises exceptions in place for Worker.start() and another to be used as a handler.

There's also a chance that I'm overlooking something and your suggestion would work in all situations, is that the case?

Functions used as handlers handle exceptions in the same way that this function is doing manually. I recommend trying things, seeing if everything passes, and reporting back.

Sure, I did that now. What I wanted to prevent is breaking some untested behavior, just like this isn't being tested currently I believe that may break some other use case out there.

Ah, perhaps we could add a small test then? I suspect that this would be the same as your current test, but you would use a pre-existing worker (just use gen_cluster rather than create things manually) and then call the client.register_plugin method (I think that that's the name but am not sure)

Thanks Matt, I really wasn't aware how to use that properly. I now added a test for register_worker_plugin, but I had to revert my previous commit, as the exception type doesn't seem to get propagated with the try/except block you suggested removing. I guess the serialization step is indeed required for handlers.

If this is still too dirty, the only idea I have right now to avoid the raise msg["exception"].data is to have two different functions for plugin_add, one that serializes the exception, and another that doesn't.

mrocklin · 2020-12-02T20:16:35Z

distributed/tests/test_worker.py

+        def setup(self, worker=None):
+            raise MyException("Foo")
+
+    s = await Scheduler(port=8007)


I recommend using `async with Scheduler(port=0) as s:

mrocklin · 2020-12-02T20:22:56Z

distributed/worker.py

            *[self.plugin_add(plugin=plugin) for plugin in self._pending_plugins]
        )
+        for msg in plugins_msgs:
+            if msg["status"] != "OK":
+                raise msg["exception"].data


I suspect that if you just remove the try-except and special handling in the plugin_add function that things may just work. I suspect that the standard machinery around rpc will handle the exception comfortably.

mrocklin · 2020-12-02T20:23:11Z

distributed/worker.py

            *[self.plugin_add(plugin=plugin) for plugin in self._pending_plugins]
        )
+        for msg in plugins_msgs:
+            if msg["status"] != "OK":
+                raise msg["exception"].data


I'm not sure though, this is just a guess. I give it a 50/50 chance.

jrbourbeau

Thanks for the PR @pentschev!

distributed/tests/test_worker.py

jrbourbeau · 2020-12-02T20:25:38Z

distributed/tests/test_worker.py

@@ -407,6 +407,29 @@ def __str__(self):
        assert "Bar" in str(e.__cause__)


+@pytest.mark.asyncio
+async def test_plugin_exception():


Suggested change

async def test_plugin_exception():

async def test_plugin_exception(cleanup):

distributed/tests/test_worker.py

pentschev · 2020-12-02T22:53:23Z

@jrbourbeau @mrocklin you both had pretty much the same comments at the same time on the tests. Responding here, the test was mostly a copy of another test, given my first attempts with async with ... had failed and the previous solution with await worked, I chose to use that one. Now with both of your suggestions everything works, all test-related changes should be in, but please let me know if I missed something.

This reverts commit cc3bec5.

pentschev · 2020-12-04T21:06:48Z

Any other suggestions on improving this PR or are we ok with its current state? The test seems unrelated too, but correct me if I'm mistaken.

pentschev · 2020-12-11T14:56:18Z

Friendly ping here, do we have anything left/other suggestions to improve this PR?

mrocklin · 2020-12-11T15:09:50Z

Sorry for the delay @pentschev . I and others have been busy recently. I would like for someone (maybe me) to take a look at why the exception isn't being passed through. I don't want our code to manually pack and unpack exceptions if we don't have to. I personally will put this on my holiday TODO list (starts at the end of next week). Hopefully I have time to look at it before then.

pentschev · 2020-12-11T15:19:26Z

No worries, thanks @mrocklin . I'm also happy to do things here, I just want to be sure on what's the general "right" way of handling such cases. Also this isn't very high-priority, so no pressure, I just wanted to make sure it wasn't forgotten. :)

…in-exceptions

…-exceptions

pentschev · 2021-11-23T22:29:10Z

@jrbourbeau @mrocklin if there are no further changes needed, could we merge this?

The failing tests are #5494 and the other pytest process seems to have died without reporting back final results in https://github.com/dask/distributed/runs/4304808476?check_suite_focus=true, but it seems unrelated to me.

quasiben · 2021-12-02T15:28:37Z

@florian-jetter-by as you've done some work on the worker recently, would you have any issue with us merging this in today or tomorrow ?

mrocklin · 2021-12-02T17:03:12Z

Pinging me these days probably isn't effective.

Also, cc'ing @fjetter (Florian's more standard username)

fjetter · 2021-12-03T11:38:57Z

distributed/worker.py

+        for msg in plugins_msgs:
+            if msg["status"] != "OK":
+                raise pickle.loads(msg["exception"].data)


I'm wondering what happens if multiple plugins raise an exception. do we at least log all of them? If not, should we iterate over all, log them and only afterwards raise?

FWIW, I think this would be a great application for exception groups proposed (and accepted) in PEP654. It's not implemented, yet, just a heads up.

Done in 7552423 .

fjetter · 2021-12-03T11:39:52Z

distributed/worker.py

+        plugins_msgs = await asyncio.gather(
            *(self.plugin_add(plugin=plugin) for plugin in self._pending_plugins)
        )


It's also possible for plugin_add to raise other exceptions (e.g. unpickling errors). What would happen in these cases? I assume it is failing hard but I would like to confirm. Do we have tests for this?

Added test for that in d323913 .

fjetter

LGTM. Once green-ish we can merge

pentschev · 2021-12-03T15:45:21Z

Failing gpuCI tests are being addressed by #5540 and #5562 .

pentschev · 2021-12-06T16:59:55Z

It seems Windows failures are unrelated to this. Failures from https://github.com/dask/distributed/runs/4409717678?check_suite_focus=true are #5494, and test_worker_reconnects_mid_compute/test_worker_reconnects_mid_compute_multiple_states_on_scheduler that don't rely on worker plugins. The other failure from https://github.com/dask/distributed/runs/4409717724?check_suite_focus=true seems like a general timeout without most tests even running. And gpuCI failures are as per #4298 (comment).

Are we good to merge this @fjetter ?

pentschev · 2022-01-12T20:13:11Z

Can we merge this @fjetter @jrbourbeau ?

gjoseph92 · 2022-01-13T20:22:56Z

distributed/worker.py

+        plugins_exceptions = []
+        for msg in plugins_msgs:
+            if msg["status"] != "OK":
+                exc = pickle.loads(msg["exception"].data)


The pickle-unpickle here feels a little cumbersome to me. I'm not sure I love this idea, but what if you add a catch_errors=True to plugin_add, and in start just set catch_errors=False?

diff --git a/distributed/worker.py b/distributed/worker.py index 6e2fc56b..f4e3ed03 100644 --- a/distributed/worker.py +++ b/distributed/worker.py @@ -1494,22 +1494,12 @@ class Worker(ServerNode): setproctitle("dask-worker [%s]" % self.address) - plugins_msgs = await asyncio.gather( - *(self.plugin_add(plugin=plugin) for plugin in self._pending_plugins) + await asyncio.gather( + *( + self.plugin_add(plugin=plugin, catch_errors=False) + for plugin in self._pending_plugins + ) ) - plugins_exceptions = [] - for msg in plugins_msgs: - if msg["status"] != "OK": - exc = pickle.loads(msg["exception"].data) - plugins_exceptions.append(pickle.loads(msg["exception"].data)) - if len(plugins_exceptions) >= 1: - if len(plugins_exceptions) > 1: - logger.error( - "Multiple plugin exceptions raised. All exceptions will be logged, the first is raised." - ) - for exc in plugins_exceptions: - logger.error(repr(exc)) - raise plugins_exceptions[0] self._pending_plugins = () @@ -3182,7 +3172,7 @@ class Worker(ServerNode): def run_coroutine(self, comm, function, args=(), kwargs=None, wait=True): return run(self, comm, function=function, args=args, kwargs=kwargs, wait=wait) - async def plugin_add(self, comm=None, plugin=None, name=None): + async def plugin_add(self, comm=None, plugin=None, name=None, catch_errors=True): with log_errors(pdb=False): if isinstance(plugin, bytes): plugin = pickle.loads(plugin) @@ -3204,6 +3194,8 @@ class Worker(ServerNode): if isawaitable(result): result = await result except Exception as e: + if not catch_errors: + raise msg = error_message(e) return msg

If you wanted to maintain the behavior you have here where multiple plugin exceptions can be handled and logged, then you could pass gather(..., return_exceptions=True) and handle them similarly as here, just without the unpickling. But I'm not sure if this functionality is actually necessary.

The pickle-unpickle here feels a little cumbersome to me. I'm not sure I love this idea, but what if you add a catch_errors=True to plugin_add, and in start just set catch_errors=False?

Thanks for the suggestion, that indeed looks better.

If you wanted to maintain the behavior you have here where multiple plugin exceptions can be handled and logged, then you could pass gather(..., return_exceptions=True) and handle them similarly as here, just without the unpickling. But I'm not sure if this functionality is actually necessary.

I'm ok with having this or not, it was suggested by @fjetter in #4298 (comment) , for now I'm keeping this but we can remove it if people prefer.

Relevant changes applied in 560bb55 .

…-exceptions

gjoseph92 · 2022-01-13T21:45:29Z

distributed/worker.py

        )
+        if len(plugins_exceptions) >= 1:


I think plugins_exceptions is not just the exceptions; it contains all the values returned by plugin_add, including OK messages. I think you'd want plugins_exceptions = [x for x in plugins_results if isinstance(x, Exception)] first.

I don't think so, that was the case with plugins_msgs (the previous pickled version), but now adding a line just above this print(f"plugins_exceptions: {plugins_exceptions}"), I see the following for distributed/tests/test_worker.py::test_plugin_exception:

plugins_exceptions: [ValueError('Setup failed')]

and for distributed/tests/test_worker.py::test_plugin_internal_exception I see:

plugins_exceptions: [UnicodeDecodeError('utf-8', b'orrupting pickle\x80\x04\x95e\x02\x00\x00\x00\x00\x00\x00\x8c\x17cloudpickle.cloudpickle\x94\x8c\r_builtin_type\x94\x93\x94\x8c', 16, 17, 'invalid start byte')]

Try having the plugin not raise an exception and see what happens.

Yes, you're absolutely right, thanks for pointing it out. This is fixed now by dc3aaeb. Could you take another look when you have a chance?

gjoseph92

LGTM besides this small nit

distributed/worker.py

Co-authored-by: Gabe Joseph <[email protected]>

pentschev · 2022-01-14T23:24:58Z

A quick look at the failing tests seem to point all failures to timeouts. Once the release is complete, can we get this merged?

quasiben · 2022-01-18T16:43:37Z

With the release completed for 2022.01.0 and the last comments addressed I'm going to merge in this PR.

Thank your @pentschev for the work and @fjetter / @gjoseph92 for the reviews

pentschev · 2022-01-18T16:51:56Z

Thanks everyone who reviewed and @quasiben for merging!

pentschev added 2 commits December 2, 2020 10:43

Raise plugin exceptions on Worker.start()

128c983

Add test for Worker plugin exceptions

3224d42

pentschev commented Dec 2, 2020

View reviewed changes

Fix black formatting

3eae880

mrocklin reviewed Dec 2, 2020

View reviewed changes

jrbourbeau reviewed Dec 2, 2020

View reviewed changes

Apply review suggestions to test_plugin_exception

fa81d3d

pentschev added 4 commits December 3, 2020 07:52

Raise exceptions in plugin_add rather then serializing them

cc3bec5

Revert "Raise exceptions in plugin_add rather then serializing them"

f85475f

This reverts commit cc3bec5.

Add test for exceptions in register_worker_plugin

dde7a09

Fix test_register_worker_plugin_exception

df4a6cb

Merge remote-tracking branch 'upstream/master' into raise-worker-plug…

51340a5

…in-exceptions

Base automatically changed from master to main March 8, 2021 19:04

pentschev added 2 commits November 23, 2021 12:50

Merge remote-tracking branch 'upstream/main' into raise-worker-plugin…

f93ab5b

…-exceptions

Unpickle plugin exceptions

5604af5

fjetter reviewed Dec 3, 2021

View reviewed changes

pentschev added 2 commits December 3, 2021 06:09

Add test for internal plugin_add exceptions

d323913

Log all plugin exceptions, raise first

7552423

fjetter approved these changes Dec 3, 2021

View reviewed changes

quasiben mentioned this pull request Jan 13, 2022

Release 2022.01.0 dask/community#211

Closed

2 tasks

gjoseph92 reviewed Jan 13, 2022

View reviewed changes

pentschev added 2 commits January 13, 2022 13:00

Merge remote-tracking branch 'upstream/main' into raise-worker-plugin…

f6a699a

…-exceptions

Raise plugin_add exceptions rather than pickling them

560bb55

gjoseph92 reviewed Jan 13, 2022

View reviewed changes

Raise only on plugin_add exceptions

dc3aaeb

gjoseph92 reviewed Jan 14, 2022

View reviewed changes

distributed/worker.py Outdated Show resolved Hide resolved

Use list comprehension to generate plugins_exceptions

d44edb4

Co-authored-by: Gabe Joseph <[email protected]>

quasiben merged commit 4c8ebfd into dask:main Jan 18, 2022

pentschev deleted the raise-worker-plugin-exceptions branch January 18, 2022 22:42

pentschev mentioned this pull request Oct 27, 2023

Failing tests on distributed>2023.9.2 rapidsai/dask-cuda#1265

Open

	async def test_plugin_exception():
	async def test_plugin_exception(cleanup):

Raise plugin exceptions on Worker.start() #4298

Raise plugin exceptions on Worker.start() #4298

Conversation

pentschev commented Dec 2, 2020

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

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

pentschev commented Dec 2, 2020

pentschev commented Dec 4, 2020

pentschev commented Dec 11, 2020

mrocklin commented Dec 11, 2020

pentschev commented Dec 11, 2020

pentschev commented Nov 23, 2021

quasiben commented Dec 2, 2021

mrocklin commented Dec 2, 2021

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

fjetter left a comment

Choose a reason for hiding this comment

pentschev commented Dec 3, 2021

pentschev commented Dec 6, 2021

pentschev commented Jan 12, 2022

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gjoseph92 left a comment

Choose a reason for hiding this comment

pentschev commented Jan 14, 2022

quasiben commented Jan 18, 2022

pentschev commented Jan 18, 2022