Close connection comm on retry

[jrbourbeau]

Currently when making new comm connections, we do so with gen.with_timeout here

distributed/distributed/comm/core.py

Lines 215 to 219 in baf5903

	comm = await gen.with_timeout(
	timedelta(seconds=min(deadline - time(), 1)),
	future,
	quiet_exceptions=EnvironmentError,
	)

and retry the connection if a timeout is reached. However, gen.with_timeout does not cancel a task when it's timeout expires. Not cancelling these tasks can lead to a "Closing dangling stream" warning being raised here

distributed/distributed/comm/tcp.py

Line 169 in baf5903

logger.warning("Closing dangling stream in %s" % (r,))

when TCP class instances from previously timed out connection task are garbage collected.

Switching to asyncio.wait_for allows us to keep a similar timeout-then-retry connection behavior, but also cancels a connection task if a timeout occurs.

This is just one proposed improvement to the situation. Alternatively we could make the currently hardcoded 1s in the timeout configurable.

[mrocklin]

This seems fine to me. Should we swap out gen.with_timeout generally throughout the codebase?

cc also @jcrist , who has been interested in tornado/asyncio switching code in the past.

[mrocklin]

Also, thanks for hunting this down @jrbourbeau !

[jcrist]

This looks good to me, thanks @jrbourbeau. Merging.

Should we swap out gen.with_timeout generally throughout the codebase?

I think we should strive for that, but because it has different semantics replacing it isn't always straightforward. Each case needs to be examined to determine what the intent was and if the lack of cancellation was accidental or necessary.

[jcrist]

Oop, just saw you marked this as [WIP]. I'll let you merge when you're happy with things. The fix looks good to me as is though.

[jrbourbeau]

Thanks for feedback @mrocklin @jcrist! Merging this now

I opened up a separate issue (#3367) for tracking gen.with_timeout -> asyncio.wait_for switches more broadly throughout the codebase