Refactor missing-data command

[crusaderky]

Partially closes #5896

[crusaderky]

all changes to scheduler.py are purely cosmetic

[github-actions]

Unit Test Results

       15 files  +       3         15 suites  +3   7h 9m 55s ⏱️ + 1h 20m 53s
  2 775 tests +       1    2 695 ✔️ +     14    79 💤  -   12  1 ❌  - 1 
20 587 runs  +3 969  19 678 ✔️ +3 856  907 💤 +113  2 ❌ ±0 

For more details on these failures, see this check.

Results for commit 6551803. ± Comparison against base commit 4b81f06.

♻️ This comment has been updated with latest results.

[gjoseph92]

For my clarification: how does worker get passed in? I don't see it as a field in MissingDataMsg.

[crusaderky]

It's automatically added by self.batched_stream.send

[gjoseph92]

I don't see logic for that anywhere in BatchedSend.send, or within the _background_send coroutine?

distributed/distributed/batched.py

Lines 131 to 143 in b8b45c6

	def send(self, *msgs) -> None:
	"""Schedule a message for sending to the other side
	This completes quickly and synchronously
	"""
	if self.comm is not None and self.comm.closed():
	raise CommClosedError(f"Comm {self.comm!r} already closed.")
	self.message_count += len(msgs)
	self.buffer.extend(msgs)
	# Avoid spurious wakeups if possible
	if self.next_deadline is None:
	self.waker.set()

[crusaderky]

It's here, on the receiving end:

distributed/distributed/scheduler.py

Line 4817 in 6e0fe58

await self.handle_stream(comm=comm, extra={"worker": worker})

[gjoseph92]

This MissingDataMsg will now come after any messages generated by transitions. I think that shouldn't matter, but want to double check.

[crusaderky]

I would be seriously worried if order mattered

[fjetter]

Ordering does sometimes matter and I suggest to not reject such a possibility just because it would be bad design. The scheduler should be able to handle all possible kind of orderings but this can still introduce subtle and unintended changes in behavior and I would advise caution either way