PYTHON-5219 - Avoid awaiting coroutines when holding locks

[NoahStapp]

No description provided.

[ShaneHarvey]

Unrelated to this PR but why do we append to _monitor_tasks here?

[NoahStapp]

To cleanup all the monitor tasks owned by the closed servers.

[ShaneHarvey]

Since asyncio doesn't support forking at all, should we just remove this? It seems like non-functional code.

[NoahStapp]

What if replace the entire fork branch here with a warning on async?

[ShaneHarvey]

Before we make any code changes someone should test out the current behavior of fork+asyncio. Depending on the behavior we might need to reopen https://jira.mongodb.org/browse/PYTHON-5249.

[NoahStapp]

This example:

import os

from pymongo import AsyncMongoClient
import asyncio

async def test_func():
    client = AsyncMongoClient()
    await client.aconnect()

    pid = os.fork()
    if pid == 0:
        await client.db.test.insert_one({'a': 1})
        exit()
    print("Done!")


asyncio.run(test_func())

Produces the following error multiple times, each with different tracebacks:

Traceback (most recent call last):
  File "/Users/nstapp/.pyenv/versions/3.13.0/lib/python3.13/asyncio/runners.py", line 194, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "/Users/nstapp/.pyenv/versions/3.13.0/lib/python3.13/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "/Users/nstapp/.pyenv/versions/3.13.0/lib/python3.13/asyncio/base_events.py", line 708, in run_until_complete
    self.run_forever()
    ~~~~~~~~~~~~~~~~^^
  File "/Users/nstapp/.pyenv/versions/3.13.0/lib/python3.13/asyncio/base_events.py", line 679, in run_forever
    self._run_once()
    ~~~~~~~~~~~~~~^^
  File "/Users/nstapp/.pyenv/versions/3.13.0/lib/python3.13/asyncio/base_events.py", line 1989, in _run_once
    event_list = self._selector.select(timeout)
  File "/Users/nstapp/.pyenv/versions/3.13.0/lib/python3.13/selectors.py", line 548, in select
    kev_list = self._selector.control(None, max_ev, timeout)
ValueError: I/O operation on closed kqueue object

[ShaneHarvey]

The same error happens without using pymongo at all:

import asyncio
import os

async def test_func():
    pid = os.fork()
    if pid == 0:
        await asyncio.sleep(.01)
        print("Done child!")
        exit()
    await asyncio.sleep(.1)
    print("Done parent!")


asyncio.run(test_func())

[ShaneHarvey]

We close the servers here but we leave self._servers untouched? Is using a client post fork broken right now? I don't see where the Server gets recreated.

[NoahStapp]

Our test_fork.py tests are all passing. We re-open each server in self._servers in _ensure_opened, called at the end of open here.

[ShaneHarvey]

Here's the race I'm concerned about:

1. app forks with an open client
2. child process starts 2 threads that both call find_one()
3. both threads see a different PID and enter this if-block.
4. T1 acquires the lock first, resets the servers, reopens and then proceeds to server selection.
5. T2 then closes the server selected by T1 which causes a PoolClosedError in T1.

It could be that this is already possible with the current code. What do you think?

[NoahStapp]

The addition of async hasn't changed the structure of this code, only the async/await syntax, so if this race condition does exist, it's existed for some time. Here's the identical PyMongo 4.8 version, before we added async support:

mongo-python-driver/pymongo/topology.py

Line 177 in de0f46a

def open(self) -> None:

I agree looking at the code that scenario certainly seems possible. We could add a flag set post-fork to prevent that race condition?

[ShaneHarvey]

Not concerned about async here. I'm concerned that delaying the server.close() call to after we release the lock will make this race more likely.

[NoahStapp]

I just realized: since we don't support fork + async at all, this change is also non-functional and can be reverted.

I mention that this code hasn't changed beside the addition of async to show that this race condition has either been present for quite some time, or doesn't exist.

[ShaneHarvey]

Yeah let's undo the forking changes. Holding the lock while calling close() in the sync version isn't so bad in this case because it only happens once post fork().