Flaky tests: coroutine 'InProc.write' was never awaited

[gjoseph92]

Catch-all issue for flaky tests with tracebacks like

    def unraisable_exception_runtest_hook() -> Generator[None, None, None]:
        with catch_unraisable_exception() as cm:
            yield
            if cm.unraisable:
                if cm.unraisable.err_msg is not None:
                    err_msg = cm.unraisable.err_msg
                else:
                    err_msg = "Exception ignored in"
                msg = f"{err_msg}: {cm.unraisable.object!r}\n\n"
                msg += "".join(
                    traceback.format_exception(
                        cm.unraisable.exc_type,
                        cm.unraisable.exc_value,
                        cm.unraisable.exc_traceback,
                    )
                )
>               warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
E               pytest.PytestUnraisableExceptionWarning: Exception ignored in: <coroutine object InProc.write at 0x7fd864183ec0>
E               
E               Traceback (most recent call last):
E                 File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/warnings.py", line 506, in _warn_unawaited_coroutine
E                   warn(msg, category=RuntimeWarning, stacklevel=2, source=coro)
E               RuntimeWarning: coroutine 'InProc.write' was never awaited

I would imagine there's some core problem that could be fixed to make all these tests work.

Some known failures (feel free to add more in comments):

• test_dont_select_closed_worker flaky #6507
• test_local_cluster_redundant_kwarg[True] flaky #6506
• test_cleanup leaked #6451
• Flaky test_run_spec #6549 (this one is interesting because it's with a TCP comm, which indicates the problem isn't with InProc but rather with comm handling)
• Flaky test_localcluster_start_exception #6769

@graingert do you have any ideas? Weren't you looking at this at some point?

[graingert]

maybe this? python/cpython#31003

[graingert]

ah no it's this

distributed/distributed/batched.py

Lines 94 to 96 in 96c13a5

	nbytes = yield self.comm.write(
	payload, serializers=self.serializers, on_error="raise"
	)

[fjetter]

xref #6389

[gjoseph92]

Would that be resolved just by switching to asyncio? I don't actually see what's causing it to not be awaited. Is it if the _background_send coroutine gets cancelled while it's awaiting self.comm.write? AFAIU _background_send actually can't be cancelled since we don't have a handle on it.

[graingert]

Would that be resolved just by switching to asyncio

no it's switching to await that fixes it here - because it moves the yield point after the Coroutine has been awaited

Is it if the _background_send coroutine gets cancelled while it's awaiting self.comm.write

No the issue is the generator getting garbage collected and raising the GeneratorExit out of the yield here rather than the Future that write goes on to yield from

[gjoseph92]

Ah, so just switching to async instead of @gen.coroutine.

[fjetter]

Is there any known issue that would stop us from switching it?

[hendrikmakait]

XREF: #6389

[gjoseph92]

@graingert and I will write up a ticket or comment on what approach we're going to take here. Refactoring BatchedSend to use asyncio right now may not help/be hard, because we don't have something to hang the long-lived task off of yet. We'll probably just ignore the warning for now.