PYTHON-1428 Respect maxPoolSize when minPoolSize is set

Author: ShaneHarvey

pymongo/pool.py

@@ -815,6 +815,7 @@ def reset(self):
  sock_info.close()
 
  def remove_stale_sockets(self):
+ """Removes stale sockets then adds new ones if pool is too small."""
  if self.opts.max_idle_time_ms is not None:
  with self.lock:
  for sock_info in self.sockets.copy():
@@ -823,11 +824,22 @@ def remove_stale_sockets(self):
  self.sockets.remove(sock_info)
  sock_info.close()
 
- while len(
- self.sockets) + self.active_sockets < self.opts.min_pool_size:
- sock_info = self.connect()
+ while True:
  with self.lock:
- self.sockets.add(sock_info)
+ if (len(self.sockets) + self.active_sockets >=
+ self.opts.min_pool_size):
+ # There are enough sockets in the pool.
+ break
+
+ # We must acquire the semaphore to respect max_pool_size.
+ if not self._socket_semaphore.acquire(False):
+ break
+ try:
+ sock_info = self.connect()
+ with self.lock:
+ self.sockets.add(sock_info)
+ finally:
+ self._socket_semaphore.release()
 
  def connect(self):
  """Connect to Mongo and return a new SocketInfo.

test/test_client.py

@@ -283,18 +283,38 @@ def test_max_idle_time_reaper(self):
  pass
  self.assertEqual(1, len(server._pool.sockets))
  self.assertTrue(sock_info in server._pool.sockets)
+ client.close()
 
  # Assert reaper removes idle socket and replaces it with a new one
  client = rs_or_single_client(maxIdleTimeMS=500,
  minPoolSize=1)
  server = client._get_topology().select_server(any_server_selector)
  with server._pool.get_socket({}) as sock_info:
  pass
+ # When the reaper runs at the same time as the get_socket, two
+ # sockets could be created and checked into the pool.
+ self.assertGreaterEqual(len(server._pool.sockets), 1)
+ wait_until(lambda: sock_info not in server._pool.sockets,
+ "remove stale socket")
+ wait_until(lambda: 1 <= len(server._pool.sockets),
+ "replace stale socket")
+ client.close()
+
+ # Assert reaper respects maxPoolSize when adding new sockets.
+ client = rs_or_single_client(maxIdleTimeMS=500,
+ minPoolSize=1,
+ maxPoolSize=1)
+ server = client._get_topology().select_server(any_server_selector)
+ with server._pool.get_socket({}) as sock_info:
+ pass
+ # When the reaper runs at the same time as the get_socket,
+ # maxPoolSize=1 should prevent two sockets from being created.
  self.assertEqual(1, len(server._pool.sockets))
  wait_until(lambda: sock_info not in server._pool.sockets,
- "reaper removes stale socket eventually")
+ "remove stale socket")
  wait_until(lambda: 1 == len(server._pool.sockets),
- "reaper replaces stale socket with new one")
+ "replace stale socket")
+ client.close()
 
  # Assert reaper has removed idle socket and NOT replaced it
  client = rs_or_single_client(maxIdleTimeMS=500)
@@ -304,6 +324,7 @@ def test_max_idle_time_reaper(self):
  wait_until(
  lambda: 0 == len(server._pool.sockets),
  "stale socket reaped and new one NOT added to the pool")
+ client.close()
 
  def test_min_pool_size(self):
  with client_knobs(kill_cursor_frequency=.1):