PYTHON-2462 Avoid connection storms: implement pool PAUSED state

Mark server unknown and clear the pool when background connections fail
Eagerly evict threads from the wait queue when pool is paused. Evicted
threads will raise the following error:
AutoReconnect('localhost:27017: connection pool paused')

CMAP spec test changes:
- CMAP unit tests should not use real monitors
- Assert that CMAP threads complete all scheduled operations

Author: ShaneHarvey

doc/changelog.rst

@@ -4,6 +4,11 @@ Changelog
 Changes in Version 4.0
 ----------------------
 
+Breaking Changes in 4.0
+```````````````````````
+
+- Removed :mod:`~pymongo.thread_util`.
+
 Issues Resolved
 ...............
 

pymongo/event_loggers.py

@@ -171,6 +171,9 @@ class ConnectionPoolLogger(monitoring.ConnectionPoolListener):
     def pool_created(self, event):
         logging.info("[pool {0.address}] pool created".format(event))
 
+    def pool_ready(self, event):
+        logging.info("[pool {0.address}] pool ready".format(event))
+
     def pool_cleared(self, event):
         logging.info("[pool {0.address}] pool cleared".format(event))
 

pymongo/mongo_client.py

@@ -737,7 +737,7 @@ def target():
 
         executor = periodic_executor.PeriodicExecutor(
             interval=common.KILL_CURSOR_FREQUENCY,
-            min_interval=0.5,
+            min_interval=common.MIN_HEARTBEAT_INTERVAL,
             target=target,
             name="pymongo_kill_cursors_thread")
 

pymongo/monitoring.py

@@ -255,6 +255,18 @@ def pool_created(self, event):
         """
         raise NotImplementedError
 
+    def pool_ready(self, event):
+        """Abstract method to handle a :class:`PoolReadyEvent`.
+
+        Emitted when a Connection Pool is marked ready.
+
+        :Parameters:
+          - `event`: An instance of :class:`PoolReadyEvent`.
+
+        .. versionadded:: 4.0
+        """
+        raise NotImplementedError
+
     def pool_cleared(self, event):
         """Abstract method to handle a `PoolClearedEvent`.
 
@@ -692,6 +704,18 @@ def __repr__(self):
             self.__class__.__name__, self.address, self.__options)
 
 
+class PoolReadyEvent(_PoolEvent):
+    """Published when a Connection Pool is marked ready.
+
+    :Parameters:
+     - `address`: The address (host, port) pair of the server this Pool is
+       attempting to connect to.
+
+    .. versionadded:: 4.0
+    """
+    __slots__ = ()
+
+
 class PoolClearedEvent(_PoolEvent):
     """Published when a Connection Pool is cleared.
 
@@ -1475,6 +1499,16 @@ def publish_pool_created(self, address, options):
             except Exception:
                 _handle_exception()
 
+    def publish_pool_ready(self, address):
+        """Publish a :class:`PoolReadyEvent` to all pool listeners.
+        """
+        event = PoolReadyEvent(address)
+        for subscriber in self.__cmap_listeners:
+            try:
+                subscriber.pool_ready(event)
+            except Exception:
+                _handle_exception()
+
     def publish_pool_cleared(self, address):
         """Publish a :class:`PoolClearedEvent` to all pool listeners.
         """

pymongo/pool.py

@@ -30,7 +30,7 @@
 from bson import DEFAULT_CODEC_OPTIONS
 from bson.py3compat import imap, itervalues, _unicode, PY3
 from bson.son import SON
-from pymongo import auth, helpers, thread_util, __version__
+from pymongo import auth, helpers, __version__
 from pymongo.client_session import _validate_session_write_concern
 from pymongo.common import (MAX_BSON_SIZE,
                             MAX_CONNECTING,
@@ -46,6 +46,7 @@
                             CertificateError,
                             ConnectionFailure,
                             ConfigurationError,
+                            ExceededMaxWaiters,
                             InvalidOperation,
                             DocumentTooLarge,
                             NetworkTimeout,
@@ -309,7 +310,8 @@ class PoolOptions(object):
                  '__wait_queue_timeout', '__wait_queue_multiple',
                  '__ssl_context', '__ssl_match_hostname', '__socket_keepalive',
                  '__event_listeners', '__appname', '__driver', '__metadata',
-                 '__compression_settings', '__max_connecting')
+                 '__compression_settings', '__max_connecting',
+                 '__pause_enabled')
 
     def __init__(self, max_pool_size=MAX_POOL_SIZE,
                  min_pool_size=MIN_POOL_SIZE,
@@ -318,7 +320,8 @@ def __init__(self, max_pool_size=MAX_POOL_SIZE,
                  wait_queue_multiple=None, ssl_context=None,
                  ssl_match_hostname=True, socket_keepalive=True,
                  event_listeners=None, appname=None, driver=None,
-                 compression_settings=None, max_connecting=MAX_CONNECTING):
+                 compression_settings=None, max_connecting=MAX_CONNECTING,
+                 pause_enabled=True):
 
         self.__max_pool_size = max_pool_size
         self.__min_pool_size = min_pool_size
@@ -335,6 +338,7 @@ def __init__(self, max_pool_size=MAX_POOL_SIZE,
         self.__driver = driver
         self.__compression_settings = compression_settings
         self.__max_connecting = max_connecting
+        self.__pause_enabled = pause_enabled
         self.__metadata = copy.deepcopy(_METADATA)
         if appname:
             self.__metadata['application'] = {'name': appname}
@@ -406,6 +410,10 @@ def max_connecting(self):
         """
         return self.__max_connecting
 
+    @property
+    def pause_enabled(self):
+        return self.__pause_enabled
+
     @property
     def max_idle_time_seconds(self):
         """The maximum number of seconds that a connection can remain
@@ -1058,6 +1066,8 @@ class _PoolClosedError(PyMongoError):
     pass
 
 
+PAUSED, READY, CLOSED = range(3)
+
 # Do *not* explicitly inherit from object or Jython won't call __del__
 # http://bugs.jython.org/issue1057
 class Pool:
@@ -1068,6 +1078,9 @@ def __init__(self, address, options, handshake=True):
           - `options`: a PoolOptions instance
           - `handshake`: whether to call ismaster for each new SocketInfo
         """
+        self.state = READY
+        if options.pause_enabled:
+            self.state = PAUSED
         # Check a socket's health with socket_closed() every once in a while.
         # Can override for testing: 0 to always check, None to never check.
         self._check_interval_seconds = 1
@@ -1079,7 +1092,6 @@ def __init__(self, address, options, handshake=True):
         self.active_sockets = 0
         # Monotonically increasing connection ID required for CMAP Events.
         self.next_connection_id = 1
-        self.closed = False
         # Track whether the sockets in this pool are writeable or not.
         self.is_writable = None
 
@@ -1098,13 +1110,23 @@ def __init__(self, address, options, handshake=True):
 
         if (self.opts.wait_queue_multiple is None or
                 self.opts.max_pool_size is None):
-            max_waiters = None
+            max_waiters = float('inf')
         else:
             max_waiters = (
                 self.opts.max_pool_size * self.opts.wait_queue_multiple)
-
-        self._socket_semaphore = thread_util.create_semaphore(
-            self.opts.max_pool_size, max_waiters)
+        # The first portion of the wait queue.
+        # Enforces: maxPoolSize and waitQueueMultiple
+        # Also used for: clearing the wait queue
+        self.size_cond = threading.Condition(self.lock)
+        self.requests = 0
+        self.max_pool_size = self.opts.max_pool_size
+        if self.max_pool_size is None:
+            self.max_pool_size = float('inf')
+        self.waiters = 0
+        self.max_waiters = max_waiters
+        # The second portion of the wait queue.
+        # Enforces: maxConnecting
+        # Also used for: clearing the wait queue
         self._max_connecting_cond = threading.Condition(self.lock)
         self._max_connecting = self.opts.max_connecting
         self._pending = 0
@@ -1114,10 +1136,22 @@ def __init__(self, address, options, handshake=True):
         # Similar to active_sockets but includes threads in the wait queue.
         self.operation_count = 0
 
+    def ready(self):
+        old_state, self.state = self.state, READY
+        if old_state != READY:
+            if self.enabled_for_cmap:
+                self.opts.event_listeners.publish_pool_ready(self.address)
+
+    @property
+    def closed(self):
+        return self.state == CLOSED
+
     def _reset(self, close):
-        with self.lock:
+        with self.size_cond:
             if self.closed:
                 return
+            if self.opts.pause_enabled:
+                self.state = PAUSED
             self.generation += 1
             newpid = os.getpid()
             if self.pid != newpid:
@@ -1126,7 +1160,10 @@ def _reset(self, close):
                 self.operation_count = 0
             sockets, self.sockets = self.sockets, collections.deque()
             if close:
-                self.closed = True
+                self.state = CLOSED
+            # Clear the wait queue
+            self._max_connecting_cond.notify_all()
+            self.size_cond.notify_all()
 
         listeners = self.opts.event_listeners
         # CMAP spec says that close() MUST close sockets before publishing the
@@ -1164,6 +1201,9 @@ def remove_stale_sockets(self, reference_generation, all_credentials):
         `generation` at the point in time this operation was requested on the
         pool.
         """
+        if self.state != READY:
+            return
+
         if self.opts.max_idle_time_seconds is not None:
             with self.lock:
                 while (self.sockets and
@@ -1172,15 +1212,14 @@ def remove_stale_sockets(self, reference_generation, all_credentials):
                     sock_info.close_socket(ConnectionClosedReason.IDLE)
 
         while True:
-            with self.lock:
+            with self.size_cond:
+                # There are enough sockets in the pool.
                 if (len(self.sockets) + self.active_sockets >=
                         self.opts.min_pool_size):
-                    # There are enough sockets in the pool.
                     return
-
-            # We must acquire the semaphore to respect max_pool_size.
-            if not self._socket_semaphore.acquire(False):
-                return
+                if self.requests >= self.opts.min_pool_size:
+                    return
+                self.requests += 1
             incremented = False
             try:
                 with self._max_connecting_cond:
@@ -1204,7 +1243,10 @@ def remove_stale_sockets(self, reference_generation, all_credentials):
                     with self._max_connecting_cond:
                         self._pending -= 1
                         self._max_connecting_cond.notify()
-                self._socket_semaphore.release()
+
+                with self.size_cond:
+                    self.requests -= 1
+                    self.size_cond.notify()
 
     def connect(self, all_credentials=None):
         """Connect to Mongo and return a new SocketInfo.
@@ -1289,6 +1331,15 @@ def get_socket(self, all_credentials, checkout=False):
             if not checkout:
                 self.return_socket(sock_info)
 
+    def _raise_if_not_ready(self):
+        if self.opts.pause_enabled and self.state == PAUSED:
+            if self.enabled_for_cmap:
+                self.opts.event_listeners.publish_connection_check_out_failed(
+                    self.address, ConnectionCheckOutFailedReason.CONN_ERROR)
+            # TODO: ensure this error is retryable
+            _raise_connection_failure(
+                self.address, AutoReconnect('connection pool paused'))
+
     def _get_socket(self, all_credentials):
         """Get or create a SocketInfo. Can raise ConnectionFailure."""
         # We use the pid here to avoid issues with fork / multiprocessing.
@@ -1313,9 +1364,26 @@ def _get_socket(self, all_credentials):
             deadline = _time() + self.opts.wait_queue_timeout
         else:
             deadline = None
-        if not self._socket_semaphore.acquire(
-                True, self.opts.wait_queue_timeout):
-            self._raise_wait_queue_timeout()
+
+        with self.size_cond:
+            self._raise_if_not_ready()
+            if self.waiters >= self.max_waiters:
+                raise ExceededMaxWaiters(
+                    'exceeded max waiters: %s threads already waiting' % (
+                        self.waiters))
+            self.waiters += 1
+            try:
+                while not (self.requests < self.max_pool_size):
+                    if not _cond_wait(self.size_cond, deadline):
+                        # Timed out, notify the next thread to ensure a
+                        # timeout doesn't consume the condition.
+                        if self.requests < self.max_pool_size:
+                            self.size_cond.notify()
+                        self._raise_wait_queue_timeout()
+                    self._raise_if_not_ready()
+            finally:
+                self.waiters -= 1
+            self.requests += 1
 
         # We've now acquired the semaphore and must release it on error.
         sock_info = None
@@ -1330,6 +1398,9 @@ def _get_socket(self, all_credentials):
                 # CMAP: we MUST wait for either maxConnecting OR for a socket
                 # to be checked back into the pool.
                 with self._max_connecting_cond:
+                    emitted_event = True
+                    self._raise_if_not_ready()
+                    emitted_event = False
                     while not (self.sockets or
                                self._pending < self._max_connecting):
                         if not _cond_wait(self._max_connecting_cond, deadline):
@@ -1340,6 +1411,9 @@ def _get_socket(self, all_credentials):
                                 self._max_connecting_cond.notify()
                             emitted_event = True
                             self._raise_wait_queue_timeout()
+                        emitted_event = True
+                        self._raise_if_not_ready()
+                        emitted_event = False
 
                     try:
                         sock_info = self.sockets.popleft()
@@ -1361,11 +1435,11 @@ def _get_socket(self, all_credentials):
             if sock_info:
                 # We checked out a socket but authentication failed.
                 sock_info.close_socket(ConnectionClosedReason.ERROR)
-            self._socket_semaphore.release()
-
-            if incremented:
-                with self.lock:
+            with self.size_cond:
+                self.requests -= 1
+                if incremented:
                     self.active_sockets -= 1
+                self.size_cond.notify()
 
             if self.enabled_for_cmap and not emitted_event:
                 self.opts.event_listeners.publish_connection_check_out_failed(
@@ -1401,10 +1475,11 @@ def return_socket(self, sock_info):
                         # Notify any threads waiting to create a connection.
                         self._max_connecting_cond.notify()
 
-        self._socket_semaphore.release()
-        with self.lock:
+        with self.size_cond:
+            self.requests -= 1
             self.active_sockets -= 1
             self.operation_count -= 1
+            self.size_cond.notify()
 
     def _perished(self, sock_info):
         """Return True and close the connection if it is "perished".