I genuinely don't understand why the asyncio.sleep avoids the broken pipe error messages, but it does.

Thanks Matt! This seems like a good change to me. 😄

Should we make an analogous change to Dask?

Should we make an analogous change to Dask?

I'm just doing this in response to user issues. I don't currently have any thoughts on what to do with dask proper. My understanding is that it isn't currently a problem, so I plan to ignore it for now.

Saw this test failure

___________________________ test_tcp_connect_timeout ___________________________
    @pytest.mark.asyncio
    async def test_tcp_connect_timeout():
>       await check_connect_timeout("tcp://127.0.0.1:44444")
distributed/comm/tests/test_comms.py:841: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
addr = 'tcp://127.0.0.1:44444'
    async def check_connect_timeout(addr):
        t1 = time()
        with pytest.raises(IOError):
>           await connect(addr, timeout=0.15)
E           Failed: DID NOT RAISE <class 'OSError'>
distributed/comm/tests/test_comms.py:834: Failed

OK, I'm going to go ahead and merge this for now. I suspect that we'll get a couple of new intermittent test failures for a bit due to the increased testing time, but it seems nicer than the alternative. cc @bnaul