- work on fixing some race-condition failures:

1. make sure pool._invalidate() sets the timestamp up before
invalidating the target connection.  we can otherwise show how the
conn.invalidate() + pool._invalidate() can lead to an extra connection
being made.
2. to help with that, soften up the check on connection.invalidate()
when connection is already closed. a warning is fine here
3. add a mutex to test_max_overflow() when we connect, because the way
we're using mock depends on an iterator, that needs to be synchronized

3 files changed, 11 insertions, 7 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 6e1564c34..9f656cac8 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -1115,8 +1115,8 @@ class Connection(Connectable):
             if self._is_disconnect:
                 del self._is_disconnect
                 dbapi_conn_wrapper = self.connection
+                self.engine.pool._invalidate(dbapi_conn_wrapper, e)
                 self.invalidate(e)
-                self.engine.pool._invalidate(dbapi_conn_wrapper)
             if self.should_close_with_result:
                 self.close()
 
diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 7fc4fc659..799443546 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -277,7 +277,7 @@ class Pool(log.Identified):
 
         return _ConnectionRecord(self)
 
-    def _invalidate(self, connection):
+    def _invalidate(self, connection, exception=None):
         """Mark all connections established within the generation
         of the given connection as invalidated.
 
@@ -291,6 +291,8 @@ class Pool(log.Identified):
         rec = getattr(connection, "_connection_record", None)
         if not rec or self._invalidate_time < rec.starttime:
             self._invalidate_time = time.time()
+        if getattr(connection, 'is_valid', False):
+            connection.invalidate(exception)
 
 
     def recreate(self):
@@ -733,7 +735,8 @@ class _ConnectionFairy(object):
         """
 
         if self.connection is None:
-            raise exc.InvalidRequestError("This connection is closed")
+            util.warn("Can't invalidate an already-closed connection.")
+            return
         if self._connection_record:
             self._connection_record.invalidate(e=e)
         self.connection = None
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index a6f40d992..a9e650cca 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -912,9 +912,11 @@ class QueuePoolTest(PoolTestBase):
         gc_collect()
 
         dbapi = MockDBAPI()
+        mutex = threading.Lock()
         def creator():
             time.sleep(.05)
-            return dbapi.connect()
+            with mutex:
+                return dbapi.connect()
 
         p = pool.QueuePool(creator=creator,
                            pool_size=3, timeout=2,
@@ -1070,7 +1072,6 @@ class QueuePoolTest(PoolTestBase):
                 # two conns
                 time.sleep(.2)
                 p._invalidate(c2)
-                c2.invalidate()
 
                 for t in threads:
                     t.join(join_timeout)
@@ -1088,7 +1089,6 @@ class QueuePoolTest(PoolTestBase):
         p1 = pool.QueuePool(creator=creator,
                            pool_size=1, timeout=None,
                            max_overflow=0)
-        #p2 = pool.NullPool(creator=creator2)
         def waiter(p):
             conn = p.connect()
             canary.append(2)
@@ -1105,7 +1105,8 @@ class QueuePoolTest(PoolTestBase):
         time.sleep(.5)
         eq_(canary, [1])
 
-        c1.invalidate()
+        # this also calls invalidate()
+        # on c1
         p1._invalidate(c1)
 
         for t in threads: