- [bug] Fixed bug whereby

a disconnect detect + dispose that occurs
when the QueuePool has threads waiting
for connections would leave those
threads waiting for the duration of
the timeout on the old pool.   The fix
now notifies those waiters with a special
exception case and has them move onto
the new pool.  This fix may or may
not be ported to 0.7.  [ticket:2522]

3 files changed, 87 insertions, 38 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index d13344ff6..75f6ac29a 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -2253,8 +2253,7 @@ class Engine(Connectable, log.Identified):
         the engine are not affected. 
 
         """
-        self.pool.dispose()
-        self.pool = self.pool.recreate()
+        self.pool = self.pool._replace()
 
     @util.deprecated("0.7", "Use the create() method on the given schema "
                             "object directly, i.e. :meth:`.Table.create`, "
diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 9aabe689b..f9e0f98dd 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -22,7 +22,6 @@ from sqlalchemy import exc, log, event, events, interfaces, util
 from sqlalchemy.util import queue as sqla_queue
 from sqlalchemy.util import threading, memoized_property, \
     chop_traceback
-
 proxies = {}
 
 def manage(module, **params):
@@ -212,6 +211,17 @@ class Pool(log.Identified):
 
         raise NotImplementedError()
 
+    def _replace(self):
+        """Dispose + recreate this pool.
+        
+        Subclasses may employ special logic to 
+        move threads waiting on this pool to the
+        new one.
+        
+        """
+        self.dispose()
+        return self.recreate()
+
     def connect(self):
         """Return a DBAPI connection from the pool.
 
@@ -580,6 +590,12 @@ class SingletonThreadPool(Pool):
             self._cleanup()
         return c
 
+class DummyLock(object):
+    def acquire(self, wait=True):
+        return True
+    def release(self):
+        pass
+
 class QueuePool(Pool):
     """A :class:`.Pool` that imposes a limit on the number of open connections.
 
@@ -688,37 +704,26 @@ class QueuePool(Pool):
         self._max_overflow = max_overflow
         self._timeout = timeout
         self._overflow_lock = self._max_overflow > -1 and \
-                                    threading.Lock() or None
-
-    def recreate(self):
-        self.logger.info("Pool recreating")
-        return self.__class__(self._creator, pool_size=self._pool.maxsize, 
-                          max_overflow=self._max_overflow,
-                          timeout=self._timeout, 
-                          recycle=self._recycle, echo=self.echo, 
-                          logging_name=self._orig_logging_name,
-                          use_threadlocal=self._use_threadlocal,
-                          _dispatch=self.dispatch)
+                                    threading.Lock() or DummyLock()
 
     def _do_return_conn(self, conn):
         try:
             self._pool.put(conn, False)
         except sqla_queue.Full:
             conn.close()
-            if self._overflow_lock is None:
+            self._overflow_lock.acquire()
+            try:
                 self._overflow -= 1
-            else:
-                self._overflow_lock.acquire()
-                try:
-                    self._overflow -= 1
-                finally:
-                    self._overflow_lock.release()
+            finally:
+                self._overflow_lock.release()
 
     def _do_get(self):
         try:
             wait = self._max_overflow > -1 and \
                         self._overflow >= self._max_overflow
             return self._pool.get(wait, self._timeout)
+        except sqla_queue.SAAbort, aborted:
+            return aborted.context._do_get()
         except sqla_queue.Empty:
             if self._max_overflow > -1 and \
                         self._overflow >= self._max_overflow:
@@ -730,22 +735,27 @@ class QueuePool(Pool):
                             "connection timed out, timeout %d" % 
                             (self.size(), self.overflow(), self._timeout))
 
-            if self._overflow_lock is not None:
-                self._overflow_lock.acquire()
-
-            if self._max_overflow > -1 and \
-                        self._overflow >= self._max_overflow:
-                if self._overflow_lock is not None:
-                    self._overflow_lock.release()
-                return self._do_get()
-
+            self._overflow_lock.acquire()
             try:
-                con = self._create_connection()
-                self._overflow += 1
+                if self._max_overflow > -1 and \
+                            self._overflow >= self._max_overflow:
+                    return self._do_get()
+                else:
+                    con = self._create_connection()
+                    self._overflow += 1
+                    return con
             finally:
-                if self._overflow_lock is not None:
-                    self._overflow_lock.release()
-            return con
+                self._overflow_lock.release()
+
+    def recreate(self):
+        self.logger.info("Pool recreating")
+        return self.__class__(self._creator, pool_size=self._pool.maxsize, 
+                          max_overflow=self._max_overflow,
+                          timeout=self._timeout, 
+                          recycle=self._recycle, echo=self.echo, 
+                          logging_name=self._orig_logging_name,
+                          use_threadlocal=self._use_threadlocal,
+                          _dispatch=self.dispatch)
 
     def dispose(self):
         while True:
@@ -758,6 +768,12 @@ class QueuePool(Pool):
         self._overflow = 0 - self.size()
         self.logger.info("Pool disposed. %s", self.status())
 
+    def _replace(self):
+        self.dispose()
+        np = self.recreate()
+        self._pool.abort(np)
+        return np
+
     def status(self):
         return "Pool size: %d  Connections in pool: %d "\
                 "Current Overflow: %d Current Checked out "\
diff --git a/lib/sqlalchemy/util/queue.py b/lib/sqlalchemy/util/queue.py
index e71ceb458..ebf736331 100644
--- a/lib/sqlalchemy/util/queue.py
+++ b/lib/sqlalchemy/util/queue.py
@@ -5,19 +5,28 @@
 # the MIT License: http://www.opensource.org/licenses/mit-license.php
 
 """An adaptation of Py2.3/2.4's Queue module which supports reentrant
-behavior, using RLock instead of Lock for its mutex object.
+behavior, using RLock instead of Lock for its mutex object.  The
+Queue object is used exclusively by the sqlalchemy.pool.QueuePool
+class.
 
 This is to support the connection pool's usage of weakref callbacks to return
 connections to the underlying Queue, which can in extremely
 rare cases be invoked within the ``get()`` method of the Queue itself,
 producing a ``put()`` inside the ``get()`` and therefore a reentrant
-condition."""
+condition.
+
+An additional change includes a special "abort" method which can be used
+to immediately raise a special exception for threads that are blocking
+on get().  This is to accommodate a rare race condition that can occur
+within QueuePool.
+
+"""
 
 from collections import deque
 from time import time as _time
 from sqlalchemy.util import threading
 
-__all__ = ['Empty', 'Full', 'Queue']
+__all__ = ['Empty', 'Full', 'Queue', 'SAAbort']
 
 class Empty(Exception):
     "Exception raised by Queue.get(block=0)/get_nowait()."
@@ -29,6 +38,11 @@ class Full(Exception):
 
     pass
 
+class SAAbort(Exception):
+    "Special SQLA exception to abort waiting"
+    def __init__(self, context):
+        self.context = context
+
 class Queue:
     def __init__(self, maxsize=0):
         """Initialize a queue object with a given maximum size.
@@ -49,6 +63,9 @@ class Queue:
         # a thread waiting to put is notified then.
         self.not_full = threading.Condition(self.mutex)
 
+        # when this is set, SAAbort is raised within get().
+        self._sqla_abort_context = False
+
     def qsize(self):
         """Return the approximate size of the queue (not reliable!)."""
 
@@ -138,6 +155,8 @@ class Queue:
             elif timeout is None:
                 while self._empty():
                     self.not_empty.wait()
+                    if self._sqla_abort_context:
+                        raise SAAbort(self._sqla_abort_context)
             else:
                 if timeout < 0:
                     raise ValueError("'timeout' must be a positive number")
@@ -147,12 +166,27 @@ class Queue:
                     if remaining <= 0.0:
                         raise Empty
                     self.not_empty.wait(remaining)
+                    if self._sqla_abort_context:
+                        raise SAAbort(self._sqla_abort_context)
             item = self._get()
             self.not_full.notify()
             return item
         finally:
             self.not_empty.release()
 
+    def abort(self, context):
+        """Issue an 'abort', will force any thread waiting on get()
+        to stop waiting and raise SAAbort.
+        
+        """
+        self._sqla_abort_context = context
+        if not self.not_full.acquire(False):
+            return
+        try:
+            self.not_empty.notify()
+        finally:
+            self.not_full.release()
+
     def get_nowait(self):
         """Remove and return an item from the queue without blocking.