- [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]

1 files changed, 63 insertions, 0 deletions

diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 789324445..4d5572891 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -802,7 +802,70 @@ class QueuePoolTest(PoolTestBase):
 
         lazy_gc()
         assert not pool._refs
+
+    def test_waiters_handled(self):
+        """test that threads waiting for connections are
+        handled when the pool is replaced.
+        
+        """
+        dbapi = MockDBAPI()
+        def creator():
+            return dbapi.connect()
  
+        success = []
+        for timeout in (None, 30):
+            for max_overflow in (-1, 0, 3):
+                p = pool.QueuePool(creator=creator,
+                                   pool_size=2, timeout=timeout,
+                                   max_overflow=max_overflow)
+                def waiter(p):
+                    conn = p.connect()
+                    time.sleep(.5)
+                    success.append(True)
+                    conn.close()
+
+                time.sleep(.2)
+                c1 = p.connect()
+                c2 = p.connect()
+
+                t = threading.Thread(target=waiter, args=(p, ))
+                t.setDaemon(True) # so the tests dont hang if this fails
+                t.start()
+
+                c1.invalidate()
+                c2.invalidate()
+                p2 = p._replace()
+        time.sleep(1)
+        eq_(len(success), 6)
+
+    def test_dispose_closes_pooled(self):
+        dbapi = MockDBAPI()
+        def creator():
+            return dbapi.connect()
+ 
+        p = pool.QueuePool(creator=creator,
+                           pool_size=2, timeout=None,
+                           max_overflow=0)
+        c1 = p.connect()
+        c2 = p.connect()
+        conns = [c1.connection, c2.connection]
+        c1.close()
+        eq_([c.closed for c in conns], [False, False])
+        p.dispose()
+        eq_([c.closed for c in conns], [True, False])
+
+        # currently, if a ConnectionFairy is closed
+        # after the pool has been disposed, there's no
+        # flag that states it should be invalidated
+        # immediately - it just gets returned to the
+        # pool normally...
+        c2.close()
+        eq_([c.closed for c in conns], [True, False])
+
+        # ...and that's the one we'll get back next.
+        c3 = p.connect()
+        assert c3.connection is conns[1]
+
     def test_no_overflow(self):
         self._test_overflow(40, 0)