make ConnectionPool.remove cancel connection attempts

[graingert]

part of #6790 (comment)

• Tests added / passed
• Passes pre-commit run --all-files

[github-actions]

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

       24 files  ±  0         24 suites  ±0   10h 39m 2s ⏱️ - 19m 34s
  3 337 tests +  1    3 235 ✔️ +  6     102 💤  - 1  0 ❌  - 4 
39 343 runs  +12  37 490 ✔️ +17  1 853 💤  - 1  0 ❌  - 4 

Results for commit 9ddee12. ± Comparison against base commit fef78d4.

♻️ This comment has been updated with latest results.

[fjetter]

Instead of mocking, you could also just connect to a server with either a very slow or blocked listener or one that is not replying on the handshake

[graingert]

I thought mocking was cleaner here, I'll look into using a slow server

[graingert]

also the other tests monkeypatch with monkeypatch.setitem(backends, "tcp", SlowBackend()), so I still think using mock.patch on the connect function is cleanest here

[graingert]

when testing this:

@gen_test()
async def test_remove_cancels_connect_attempts():
    loop = asyncio.get_running_loop()
    connect_started = asyncio.Event()
    connect_finished = asyncio.Event()

    class BrokenHandshakeListener(TCPListener):
        async def on_connection(self, comm):
            try:
                connect_started.set()
                await comm.read(1)
            finally:
                connect_finished.set()

    async with BrokenHandshakeListener(
        address="tcp://",
        comm_handler=lambda: None,
    ) as listener:
        rpc = await ConnectionPool(limit=1)

        async def connect_to_server():
            with pytest.raises(CommClosedError, match="Address removed."):
                await rpc.connect(listener.contact_address)

        async def remove_address():
            await connect_started.wait()
            rpc.remove(listener.contact_address)

        await asyncio.gather(
            connect_to_server(),
            remove_address(),
        )

        await connect_finished.wait()

I hit a race condition in this asyncio.wait_for where the cancellation is ignored:

distributed/distributed/comm/core.py

Lines 291 to 295 in 9ddee12

	comm = await asyncio.wait_for(
	connector.connect(loc, deserialize=deserialize, **connection_args),
	timeout=min(intermediate_cap, time_left()),
	)
	break

https://github.com/python/cpython/blob/924a3bfa28578802eb9ca77a66fb5d4762a62f14/Lib/asyncio/tasks.py#L472

this is because the connector.connect and on_connection tasks resume in the same even loop cycle and so the cancellation arrives just as the connect() -> asyncio.wait_for() coroutine is about to be resumed, this is very unlikely in production because the code will be waiting in comm.read() or asyncio.sleep(backoff) and can be resolved in the test by adding an asyncio.sleep(0.5) before calling rpc.remove():

        async def remove_address():
            await connect_started.wait()
            await asyncio.sleep(0.5)  # avoid issuing a .cancel() after connect but before comm.read()
            rpc.remove(listener.contact_address)

or do you think it's best to leave this with a mocked connect function?

[fjetter]

I think we should come back to this once #7571 is done. Triggering this edge case is interesting and I believe we've encountered this a couple of times in the past (in CI).

No need for any action on this PR.

[fjetter]

I think we should come back to this once #7571 is done. Triggering this edge case is interesting and I believe we've encountered this a couple of times in the past (in CI).

No need for any action on this PR.