Skip to content

PYTHON-5212 - Do not hold Topology lock while resetting pool #2301

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
+52 −24
Open

PYTHON-5212 - Do not hold Topology lock while resetting pool #2301

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
216407e
PYTHON-5212 - Do not hold Topology lock while resetting pool
NoahStapp Apr 21, 2025
4f0c8e4
Use correct close_conn
NoahStapp Apr 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 22 additions & 6 deletions pymongo/asynchronous/pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

from __future__ import annotations

import asyncio
import collections
import contextlib
import logging
Expand Down Expand Up @@ -860,8 +861,13 @@ async def _reset(
# PoolClosedEvent but that reset() SHOULD close sockets *after*
# publishing the PoolClearedEvent.
if close:
for conn in sockets:
await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
if not _IS_SYNC:
await asyncio.gather(
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We probably want to use return_exceptions=True here to ensure all tasks complete.

*[conn.close_conn(ConnectionClosedReason.POOL_CLOSED) for conn in sockets]
)
else:
for conn in sockets:
await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_closed(self.address)
Expand Down Expand Up @@ -891,8 +897,13 @@ async def _reset(
serverPort=self.address[1],
serviceId=service_id,
)
for conn in sockets:
await conn.close_conn(ConnectionClosedReason.STALE)
if not _IS_SYNC:
await asyncio.gather(
*[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets]
)
else:
for conn in sockets:
await conn.close_conn(ConnectionClosedReason.STALE)

async def update_is_writable(self, is_writable: Optional[bool]) -> None:
"""Updates the is_writable attribute on all sockets currently in the
Expand Down Expand Up @@ -938,8 +949,13 @@ async def remove_stale_sockets(self, reference_generation: int) -> None:
and self.conns[-1].idle_time_seconds() > self.opts.max_idle_time_seconds
):
close_conns.append(self.conns.pop())
for conn in close_conns:
await conn.close_conn(ConnectionClosedReason.IDLE)
if not _IS_SYNC:
await asyncio.gather(
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same here.

*[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns]
)
else:
for conn in close_conns:
await conn.close_conn(ConnectionClosedReason.IDLE)

while True:
async with self.size_cond:
Expand Down
11 changes: 5 additions & 6 deletions pymongo/asynchronous/topology.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -529,12 +529,6 @@ async def _process_change(
if not _IS_SYNC:
self._monitor_tasks.append(self._srv_monitor)

# Clear the pool from a failed heartbeat.
if reset_pool:
server = self._servers.get(server_description.address)
if server:
await server.pool.reset(interrupt_connections=interrupt_connections)

# Wake anything waiting in select_servers().
self._condition.notify_all()

Expand All @@ -557,6 +551,11 @@ async def on_change(
# that didn't include this server.
if self._opened and self._description.has_server(server_description.address):
await self._process_change(server_description, reset_pool, interrupt_connections)
# Clear the pool from a failed heartbeat, done outside the lock to avoid blocking on connection close.
if self._opened and self._description.has_server(server_description.address) and reset_pool:
server = self._servers.get(server_description.address)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The has_server -> _servers.get pattern is not safe to do here (https://en.wikipedia.org/wiki/Time-of-check_to_time-of-use)

We also don't need to check for closed/opened because pool.reset is safe to call even after close().

Instead we can do:

if reset_pool:
    server = self._servers.get(server_description.address)
    if server:
        await server.pool.reset(interrupt_connections=interrupt_connections)

if server:
await server.pool.reset(interrupt_connections=interrupt_connections)

async def _process_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
"""Process a new seedlist on an opened topology.
Expand Down
26 changes: 20 additions & 6 deletions pymongo/synchronous/pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

from __future__ import annotations

import asyncio
import collections
import contextlib
import logging
Expand Down Expand Up @@ -858,8 +859,13 @@ def _reset(
# PoolClosedEvent but that reset() SHOULD close sockets *after*
# publishing the PoolClearedEvent.
if close:
for conn in sockets:
conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
if not _IS_SYNC:
asyncio.gather(
*[conn.close_conn(ConnectionClosedReason.POOL_CLOSED) for conn in sockets]
)
else:
for conn in sockets:
conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_closed(self.address)
Expand Down Expand Up @@ -889,8 +895,11 @@ def _reset(
serverPort=self.address[1],
serviceId=service_id,
)
for conn in sockets:
conn.close_conn(ConnectionClosedReason.STALE)
if not _IS_SYNC:
asyncio.gather(*[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets])
else:
for conn in sockets:
conn.close_conn(ConnectionClosedReason.STALE)

def update_is_writable(self, is_writable: Optional[bool]) -> None:
"""Updates the is_writable attribute on all sockets currently in the
Expand Down Expand Up @@ -934,8 +943,13 @@ def remove_stale_sockets(self, reference_generation: int) -> None:
and self.conns[-1].idle_time_seconds() > self.opts.max_idle_time_seconds
):
close_conns.append(self.conns.pop())
for conn in close_conns:
conn.close_conn(ConnectionClosedReason.IDLE)
if not _IS_SYNC:
asyncio.gather(
*[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns]
)
else:
for conn in close_conns:
conn.close_conn(ConnectionClosedReason.IDLE)

while True:
with self.size_cond:
Expand Down
11 changes: 5 additions & 6 deletions pymongo/synchronous/topology.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -529,12 +529,6 @@ def _process_change(
if not _IS_SYNC:
self._monitor_tasks.append(self._srv_monitor)

# Clear the pool from a failed heartbeat.
if reset_pool:
server = self._servers.get(server_description.address)
if server:
server.pool.reset(interrupt_connections=interrupt_connections)

# Wake anything waiting in select_servers().
self._condition.notify_all()

Expand All @@ -557,6 +551,11 @@ def on_change(
# that didn't include this server.
if self._opened and self._description.has_server(server_description.address):
self._process_change(server_description, reset_pool, interrupt_connections)
# Clear the pool from a failed heartbeat, done outside the lock to avoid blocking on connection close.
if self._opened and self._description.has_server(server_description.address) and reset_pool:
server = self._servers.get(server_description.address)
if server:
server.pool.reset(interrupt_connections=interrupt_connections)

def _process_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
"""Process a new seedlist on an opened topology.
Expand Down
Loading