fix handling of maxConnecting

Author: blink1073

pymongo/asynchronous/pool.py

@@ -788,7 +788,6 @@ def __init__(
         # Enforces: maxConnecting
         # Also used for: clearing the wait queue
         self._max_connecting_cond = _async_create_condition(self.lock)
-        self._max_connecting = self.opts.max_connecting
         self._pending = 0
         self._client_id = client_id
         self._backoff = 0
@@ -931,6 +930,11 @@ async def _reset(
                 for conn in sockets:
                     await conn.close_conn(ConnectionClosedReason.STALE)
 
+    @property
+    def max_connecting(self) -> int:
+        """The current max connecting limit for the pool."""
+        return 1 if self._backoff else self.opts.max_connecting
+
     async def update_is_writable(self, is_writable: Optional[bool]) -> None:
         """Updates the is_writable attribute on all sockets currently in the
         Pool.
@@ -997,8 +1001,7 @@ async def remove_stale_sockets(self, reference_generation: int) -> None:
                 async with self._max_connecting_cond:
                     # If maxConnecting connections are already being created
                     # by this pool then try again later instead of waiting.
-                    max_connecting = 1 if self._backoff else self._max_connecting
-                    if self._pending >= max_connecting:
+                    if self._pending >= self.max_connecting:
                         return
                     self._pending += 1
                     incremented = True
@@ -1297,13 +1300,12 @@ async def _get_conn(
                 # to be checked back into the pool.
                 async with self._max_connecting_cond:
                     self._raise_if_not_ready(checkout_started_time, emit_event=False)
-                    max_connecting = 1 if self._backoff else self._max_connecting
-                    while not (self.conns or self._pending < max_connecting):
+                    while not (self.conns or self._pending < self.max_connecting):
                         timeout = deadline - time.monotonic() if deadline else None
                         if not await _async_cond_wait(self._max_connecting_cond, timeout):
                             # Timed out, notify the next thread to ensure a
                             # timeout doesn't consume the condition.
-                            if self.conns or self._pending < max_connecting:
+                            if self.conns or self._pending < self.max_connecting:
                                 self._max_connecting_cond.notify()
                             emitted_event = True
                             self._raise_wait_queue_timeout(checkout_started_time)

pymongo/synchronous/pool.py

@@ -786,7 +786,6 @@ def __init__(
         # Enforces: maxConnecting
         # Also used for: clearing the wait queue
         self._max_connecting_cond = _create_condition(self.lock)
-        self._max_connecting = self.opts.max_connecting
         self._pending = 0
         self._client_id = client_id
         self._backoff = 0
@@ -929,6 +928,11 @@ def _reset(
                 for conn in sockets:
                     conn.close_conn(ConnectionClosedReason.STALE)
 
+    @property
+    def max_connecting(self) -> int:
+        """The current max connecting limit for the pool."""
+        return 1 if self._backoff else self.opts.max_connecting
+
     def update_is_writable(self, is_writable: Optional[bool]) -> None:
         """Updates the is_writable attribute on all sockets currently in the
         Pool.
@@ -993,8 +997,7 @@ def remove_stale_sockets(self, reference_generation: int) -> None:
                 with self._max_connecting_cond:
                     # If maxConnecting connections are already being created
                     # by this pool then try again later instead of waiting.
-                    max_connecting = 1 if self._backoff else self._max_connecting
-                    if self._pending >= max_connecting:
+                    if self._pending >= self.max_connecting:
                         return
                     self._pending += 1
                     incremented = True
@@ -1293,13 +1296,12 @@ def _get_conn(
                 # to be checked back into the pool.
                 with self._max_connecting_cond:
                     self._raise_if_not_ready(checkout_started_time, emit_event=False)
-                    max_connecting = 1 if self._backoff else self._max_connecting
-                    while not (self.conns or self._pending < max_connecting):
+                    while not (self.conns or self._pending < self.max_connecting):
                         timeout = deadline - time.monotonic() if deadline else None
                         if not _cond_wait(self._max_connecting_cond, timeout):
                             # Timed out, notify the next thread to ensure a
                             # timeout doesn't consume the condition.
-                            if self.conns or self._pending < max_connecting:
+                            if self.conns or self._pending < self.max_connecting:
                                 self._max_connecting_cond.notify()
                             emitted_event = True
                             self._raise_wait_queue_timeout(checkout_started_time)

test/asynchronous/test_pooling.py

@@ -395,52 +395,40 @@ async def test_checkout_more_than_max_pool_size(self):
             await asyncio.sleep(0.05)
         await pool.close()
 
-    async def _check_maxConnecting(
-        self, client: AsyncMongoClient, backoff=False
-    ) -> tuple[int, int]:
-        coll = client.test.test.with_options(read_preference=ReadPreference.PRIMARY)
-        await coll.insert_one({})
-
+    async def test_maxConnecting(self):
+        client = await self.async_rs_or_single_client()
+        await self.client.test.test.insert_one({})
+        self.addAsyncCleanup(self.client.test.test.delete_many, {})
         pool = await async_get_pool(client)
-        if backoff:
-            pool._backoff = 1
         docs = []
 
         # Run 50 short running operations
         async def find_one():
-            docs.append(await coll.find_one({}))
+            docs.append(await client.test.test.find_one({}))
 
         tasks = [ConcurrentRunner(target=find_one) for _ in range(50)]
         for task in tasks:
             await task.start()
         for task in tasks:
             await task.join(10)
 
-        await coll.delete_many({})
-
-        return len(docs), len(pool.conns)
-
-    async def test_maxConnecting(self):
-        client = await self.async_rs_or_single_client()
-        num_docs, num_conns = await self._check_maxConnecting(client)
-
-        self.assertEqual(num_docs, 50)
-        self.assertLessEqual(num_conns, 50)
+        self.assertEqual(len(docs), 50)
+        self.assertLessEqual(len(pool.conns), 50)
         # TLS and auth make connection establishment more expensive than
         # the query which leads to more threads hitting maxConnecting.
         # The end result is fewer total connections and better latency.
         if async_client_context.tls and async_client_context.auth_enabled:
-            self.assertLessEqual(num_conns, 30)
+            self.assertLessEqual(len(pool.conns), 30)
         else:
-            self.assertLessEqual(num_conns, 50)
+            self.assertLessEqual(len(pool.conns), 50)
         # MongoDB 4.4.1 with auth + ssl:
         # maxConnecting = 2:         6 connections in ~0.231+ seconds
         # maxConnecting = unbounded: 50 connections in ~0.642+ seconds
         #
         # MongoDB 4.4.1 with no-auth no-ssl Python 3.8:
         # maxConnecting = 2:         15-22 connections in ~0.108+ seconds
         # maxConnecting = unbounded: 30+ connections in ~0.140+ seconds
-        print(num_conns)
+        print(len(pool.conns))
 
     @async_client_context.require_failCommand_appName
     async def test_csot_timeout_message(self):
@@ -588,17 +576,13 @@ async def test_pool_backoff_preserves_existing_connections(self):
         await pool.close()
 
     async def test_pool_backoff_limits_maxConnecting(self):
-        client = await self.async_rs_or_single_client(maxConnecting=30)
-        _, baseline_conns = await self._check_maxConnecting(client)
-        await client.close()
-
-        client = await self.async_rs_or_single_client(maxConnecting=30)
-        _, backoff_conns = await self._check_maxConnecting(client, backoff=True)
+        client = await self.async_rs_or_single_client(maxConnecting=10)
+        pool = await async_get_pool(client)
+        assert pool.maxConnecting == 10
+        pool._backoff = 1
+        assert pool.maxConnecting == 1
         await client.close()
 
-        # We should have created less conns due to limiting maxConnecting.
-        self.assertLess(backoff_conns, baseline_conns)
-
 
 class TestPoolMaxSize(_TestPoolingBase):
     async def test_max_pool_size(self):

test/test_pooling.py

@@ -395,50 +395,40 @@ def test_checkout_more_than_max_pool_size(self):
             time.sleep(0.05)
         pool.close()
 
-    def _check_maxConnecting(self, client: MongoClient, backoff=False) -> tuple[int, int]:
-        coll = client.test.test.with_options(read_preference=ReadPreference.PRIMARY)
-        coll.insert_one({})
-
+    def test_maxConnecting(self):
+        client = self.rs_or_single_client()
+        self.client.test.test.insert_one({})
+        self.addCleanup(self.client.test.test.delete_many, {})
         pool = get_pool(client)
-        if backoff:
-            pool._backoff = 1
         docs = []
 
         # Run 50 short running operations
         def find_one():
-            docs.append(coll.find_one({}))
+            docs.append(client.test.test.find_one({}))
 
         tasks = [ConcurrentRunner(target=find_one) for _ in range(50)]
         for task in tasks:
             task.start()
         for task in tasks:
             task.join(10)
 
-        coll.delete_many({})
-
-        return len(docs), len(pool.conns)
-
-    def test_maxConnecting(self):
-        client = self.rs_or_single_client()
-        num_docs, num_conns = self._check_maxConnecting(client)
-
-        self.assertEqual(num_docs, 50)
-        self.assertLessEqual(num_conns, 50)
+        self.assertEqual(len(docs), 50)
+        self.assertLessEqual(len(pool.conns), 50)
         # TLS and auth make connection establishment more expensive than
         # the query which leads to more threads hitting maxConnecting.
         # The end result is fewer total connections and better latency.
         if client_context.tls and client_context.auth_enabled:
-            self.assertLessEqual(num_conns, 30)
+            self.assertLessEqual(len(pool.conns), 30)
         else:
-            self.assertLessEqual(num_conns, 50)
+            self.assertLessEqual(len(pool.conns), 50)
         # MongoDB 4.4.1 with auth + ssl:
         # maxConnecting = 2:         6 connections in ~0.231+ seconds
         # maxConnecting = unbounded: 50 connections in ~0.642+ seconds
         #
         # MongoDB 4.4.1 with no-auth no-ssl Python 3.8:
         # maxConnecting = 2:         15-22 connections in ~0.108+ seconds
         # maxConnecting = unbounded: 30+ connections in ~0.140+ seconds
-        print(num_conns)
+        print(len(pool.conns))
 
     @client_context.require_failCommand_appName
     def test_csot_timeout_message(self):
@@ -584,17 +574,13 @@ def test_pool_backoff_preserves_existing_connections(self):
         pool.close()
 
     def test_pool_backoff_limits_maxConnecting(self):
-        client = self.rs_or_single_client(maxConnecting=30)
-        _, baseline_conns = self._check_maxConnecting(client)
-        client.close()
-
-        client = self.rs_or_single_client(maxConnecting=30)
-        _, backoff_conns = self._check_maxConnecting(client, backoff=True)
+        client = self.rs_or_single_client(maxConnecting=10)
+        pool = get_pool(client)
+        assert pool.maxConnecting == 10
+        pool._backoff = 1
+        assert pool.maxConnecting == 1
         client.close()
 
-        # We should have created less conns due to limiting maxConnecting.
-        self.assertLess(backoff_conns, baseline_conns)
-
 
 class TestPoolMaxSize(_TestPoolingBase):
     def test_max_pool_size(self):