Client story by sjperkins · Pull Request #5987 · dask/distributed

sjperkins · 2022-03-24T09:02:31Z

Closes Client.story - Support collecting cluster-wide story for a key or stimulus ID #5872
Tests added / passed
Passes pre-commit run --all-files

…| None'

sjperkins · 2022-03-24T09:08:04Z

This PR is based on top of #5954 and should be rebased when #5954 is merged.

TODO:

At present, assert_story's monotonically increasing timestamp requirement doesn't work well the cluster-wide stories.
The Client currently requests the Scheduler for story keys, which in turn requests stories from Workers. It may be possible for the Client to simply request stories from the Workers.
Comm errors when requesting stories are not gracefully handled.

github-actions · 2022-03-24T09:14:57Z

Unit Test Results

      12 files ±  0       12 suites ±0 6h 19m 36s ⏱️ - 16m 38s
  2 673 tests +  4   2 591 ✔️ +  4   81 💤 - 1 1 ❌ +1
15 942 runs +24 15 077 ✔️ +22 859 💤 - 4 6 ❌ +6

For more details on these failures, see this check.

Results for commit 2888b26. ± Comparison against base commit 6dd928b.

♻️ This comment has been updated with latest results.

sjperkins · 2022-03-24T09:30:24Z

From #5872

The failure behaviour of timed out workers should be configurable (raise, ignore). TBD: Should there be a sentinel if ignored?

and #5987 (comment)

Comm errors when requesting stories are not gracefully handled.

asyncio.gather(..., return_exceptions=True) offers a way forward.

graingert · 2022-03-24T16:05:51Z

distributed/tests/test_client.py

+
+    async def get_story(self, *args, **kw):
+        await self.unblock_worker.wait()
+        return super().get_story(*args, **kw)


Suggested change

return super().get_story(*args, **kw)

return await super().get_story(*args, **kw)

graingert · 2022-03-24T16:07:59Z

distributed/client.py

+            return await task
+        except Exception:
+            if on_error == "raise":
+                task.cancel()


the task will always be done at this point - because of the return await task

graingert · 2022-03-24T16:26:56Z

distributed/scheduler.py

+        bits = [
+            (ws, await self.rpc.connect(ws.address)) for ws in self.workers.values()
+        ]
+        tasks = []
+
+        for _, worker_comm in bits:
+            coro = send_recv(comm=worker_comm, reply=True, op="get_story", keys=keys)
+            tasks.append(asyncio.ensure_future(coro))
+
+        try:
+            worker_stories = await asyncio.gather(
+                *tasks, return_exceptions=on_error == "ignore"
+            )
+        except Exception:
+            for task in tasks:
+                task.cancel()
+
+            raise
+        finally:
+            for worker_state, worker_comm in bits:
+                self.rpc.reuse(worker_state.address, worker_comm)


I think reproducing the pattern from

distributed/distributed/scheduler.py

Lines 6155 to 6187 in 5c7d555

async def send_message(addr):

try:

comm = await self.rpc.connect(addr)

comm.name = "Scheduler Broadcast"

try:

resp = await send_recv(

comm, close=True, serializers=serializers, **msg

)

finally:

self.rpc.reuse(addr, comm)

return resp

except Exception as e:

logger.error(f"broadcast to {addr} failed: {e.__class__.__name__}: {e}")

if on_error == "raise":

raise

elif on_error == "return":

return e

elif on_error == "return_pickle":

return dumps(e, protocol=4)

elif on_error == "ignore":

return ERROR

else:

raise ValueError(

"on_error must be 'raise', 'return', 'return_pickle', "

f"or 'ignore'; got {on_error!r}"

)

results = await All(

[send_message(address) for address in addresses if address is not None]

)

return {k: v for k, v in zip(workers, results) if v is not ERROR}

would be cleaner than using gather/return_exceptions/cancel as they all interact slightly strangely

This reverts commit 925c2bd.

…erialisation on exception

…ctly from the workers

fjetter · 2022-05-17T13:57:03Z

Closed by #6161

sjperkins added 19 commits March 16, 2022 17:21

Initial commit

2957f38

Defer to parent stimulus in transition_processing_memory

e2f54a3

Change exception_blame type to TaskState | None

5307c0e

Change exception_blame back to TaskState and make stimulus_id's 'str …

c229a65

…| None'

Remove stimulus_id type annotations

d010b2d

Add stimulus_id's in various places

6cfc1ef

Merge branch 'main' into scheduler-task-transition-tracing

2ced7c1

Add stimulus_id's in state mutating handlers

cb15e74

Reintroduce log handlers

21302bd

Merge branch 'main' into scheduler-task-transition-tracing

2cf4a8a

Checkpoint running test suite

1670906

Remove cruft

8c577f8

Add stimulus_id to scheduler transition_log

dabfefe

Support worker + client stimuli, stimuli in http

be43e63

assert_worker_story -> assert_story and test scheduler story

45bf7de

Remove stimulus_id's from code paths that don't change TaskState

80ffddf

Merge branch 'main' into scheduler-task-transition-tracing

a246197

Further stimulus_id changes in worker.py

1a72c88

Add missing stimulus_id to free-keys worker msg

7c30cfa

sjperkins marked this pull request as draft March 24, 2022 09:02

sjperkins self-assigned this Mar 24, 2022

graingert reviewed Mar 24, 2022

View reviewed changes

sjperkins added 3 commits March 25, 2022 16:01

Review changes

b0fd952

Merge branch 'main' into scheduler-task-transition-tracing

90039fb

Fix stray assert_worker_story -> assert_story

3affd07

sjperkins added 8 commits March 25, 2022 18:16

Used new_stimulus_id in Worker.execute

925c2bd

Revert "Used new_stimulus_id in Worker.execute"

6e58f91

This reverts commit 925c2bd.

Initial implemention of client story

aa218b0

Handle raise and ignore on scheduler/worker comm errors

fd6e155

Try using cancellable asyncio tasks in attempt to address coroutine s…

b12bcca

…erialisation on exception

Refactor send_message into utils_comm.py

c3aa37e

Refactor to use send_message on client, and get the worker story dire…

f57ddb6

…ctly from the workers

nits

2888b26

sjperkins force-pushed the client-story branch from 056c077 to 2888b26 Compare March 25, 2022 16:48

sjperkins mentioned this pull request Apr 12, 2022

Support Stimulus ID's via argument passing #6095

Closed

4 tasks

fjetter closed this May 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Client story#5987

Client story#5987
sjperkins wants to merge 30 commits intodask:mainfrom
sjperkins:client-story

sjperkins commented Mar 24, 2022 •

edited

Loading

Uh oh!

sjperkins commented Mar 24, 2022

Uh oh!

github-actions bot commented Mar 24, 2022 •

edited

Loading

Uh oh!

sjperkins commented Mar 24, 2022 •

edited

Loading

Uh oh!

graingert Mar 24, 2022

Uh oh!

graingert Mar 24, 2022 •

edited

Loading

Uh oh!

graingert Mar 24, 2022 •

edited

Loading

Uh oh!

fjetter commented May 17, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	return super().get_story(args, *kw)
	return await super().get_story(args, *kw)


	async def send_message(addr):
	try:
	comm = await self.rpc.connect(addr)
	comm.name = "Scheduler Broadcast"
	try:
	resp = await send_recv(
	comm, close=True, serializers=serializers, **msg
	)
	finally:
	self.rpc.reuse(addr, comm)
	return resp
	except Exception as e:
	logger.error(f"broadcast to {addr} failed: {e.__class__.__name__}: {e}")
	if on_error == "raise":
	raise
	elif on_error == "return":
	return e
	elif on_error == "return_pickle":
	return dumps(e, protocol=4)
	elif on_error == "ignore":
	return ERROR
	else:
	raise ValueError(
	"on_error must be 'raise', 'return', 'return_pickle', "
	f"or 'ignore'; got {on_error!r}"
	)

	results = await All(
	[send_message(address) for address in addresses if address is not None]
	)

	return {k: v for k, v in zip(workers, results) if v is not ERROR}

Uh oh!

Conversation

sjperkins commented Mar 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sjperkins commented Mar 24, 2022

Uh oh!

github-actions bot commented Mar 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Unit Test Results

Uh oh!

sjperkins commented Mar 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

graingert Mar 24, 2022

Choose a reason for hiding this comment

Uh oh!

graingert Mar 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

graingert Mar 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fjetter commented May 17, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

sjperkins commented Mar 24, 2022 •

edited

Loading

github-actions bot commented Mar 24, 2022 •

edited

Loading

sjperkins commented Mar 24, 2022 •

edited

Loading

graingert Mar 24, 2022 •

edited

Loading

graingert Mar 24, 2022 •

edited

Loading