Create explicit GC ordering between ConnectionFairy/ConnectionRecord

Fixed issue where connection pool would not return connections to the pool
or otherwise be finalized upon garbage collection under pypy if the checked
out connection fell out of scope without being closed.   This is a long
standing issue due to pypy's difference in GC behavior that does not call
weakref finalizers if they are relative to another object that is also
being garbage collected.  A strong reference to the related record is now
maintained so that the weakref has a strong-referenced "base" to trigger
off of.

Fixes: #5842
Change-Id: Id5448fdacb6cceaac1ea40b2fbc851f052ed8e86

4 files changed, 59 insertions, 17 deletions

diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index 6c3aad037..47d9e2cba 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -18,7 +18,6 @@ from .. import event
 from .. import exc
 from .. import log
 from .. import util
-from ..util import threading
 
 
 reset_rollback = util.symbol("reset_rollback")
@@ -172,7 +171,6 @@ class Pool(log.Identified):
             self._orig_logging_name = None
 
         log.instance_logger(self, echoflag=echo)
-        self._threadconns = threading.local()
         self._creator = creator
         self._recycle = recycle
         self._invalidate_time = 0
@@ -423,27 +421,37 @@ class _ConnectionRecord(object):
             dbapi_connection = rec.get_connection()
         except Exception as err:
             with util.safe_reraise():
-                rec._checkin_failed(err)
+                rec._checkin_failed(err, _fairy_was_created=False)
         echo = pool._should_log_debug()
         fairy = _ConnectionFairy(dbapi_connection, rec, echo)
 
-        rec.fairy_ref = weakref.ref(
+        rec.fairy_ref = ref = weakref.ref(
             fairy,
             lambda ref: _finalize_fairy
             and _finalize_fairy(None, rec, pool, ref, echo),
         )
+        _strong_ref_connection_records[ref] = rec
         if echo:
             pool.logger.debug(
                 "Connection %r checked out from pool", dbapi_connection
             )
         return fairy
 
-    def _checkin_failed(self, err):
+    def _checkin_failed(self, err, _fairy_was_created=True):
         self.invalidate(e=err)
-        self.checkin(_no_fairy_ref=True)
+        self.checkin(
+            _fairy_was_created=_fairy_was_created,
+        )
 
-    def checkin(self, _no_fairy_ref=False):
-        if self.fairy_ref is None and not _no_fairy_ref:
+    def checkin(self, _fairy_was_created=True):
+        if self.fairy_ref is None and _fairy_was_created:
+            # _fairy_was_created is False for the initial get connection phase;
+            # meaning there was no _ConnectionFairy and we must unconditionally
+            # do a checkin.
+            #
+            # otherwise, if fairy_was_created==True, if fairy_ref is None here
+            # that means we were checked in already, so this looks like
+            # a double checkin.
             util.warn("Double checkin attempted on %s" % self)
             return
         self.fairy_ref = None
@@ -454,6 +462,7 @@ class _ConnectionRecord(object):
             finalizer(connection)
         if pool.dispatch.checkin:
             pool.dispatch.checkin(connection, self)
+
         pool._return_conn(self)
 
     @property
@@ -604,6 +613,11 @@ def _finalize_fairy(
 
     """
 
+    if ref:
+        _strong_ref_connection_records.pop(ref, None)
+    elif fairy:
+        _strong_ref_connection_records.pop(weakref.ref(fairy), None)
+
     if ref is not None:
         if connection_record.fairy_ref is not ref:
             return
@@ -663,6 +677,13 @@ def _finalize_fairy(
         connection_record.checkin()
 
 
+# a dictionary of the _ConnectionFairy weakrefs to _ConnectionRecord, so that
+# GC under pypy will call ConnectionFairy finalizers.  linked directly to the
+# weakref that will empty itself when collected so that it should not create
+# any unmanaged memory references.
+_strong_ref_connection_records = {}
+
+
 class _ConnectionFairy(object):
 
     """Proxies a DBAPI connection and provides return-on-dereference
@@ -797,7 +818,17 @@ class _ConnectionFairy(object):
                     )
                 except Exception as err:
                     with util.safe_reraise():
-                        fairy._connection_record._checkin_failed(err)
+                        fairy._connection_record._checkin_failed(
+                            err,
+                            _fairy_was_created=True,
+                        )
+
+                        # prevent _ConnectionFairy from being carried
+                        # in the stack trace.  Do this after the
+                        # connection record has been checked in, so that
+                        # if the del triggers a finalize fairy, it won't
+                        # try to checkin a second time.
+                        del fairy
 
                 attempts -= 1
 
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index 40549f54c..f2ed91b79 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -228,6 +228,10 @@ def is_none(a, msg=None):
     is_(a, None, msg=msg)
 
 
+def is_not_none(a, msg=None):
+    is_not(a, None, msg=msg)
+
+
 def is_true(a, msg=None):
     is_(bool(a), True, msg=msg)
 
@@ -236,14 +240,6 @@ def is_false(a, msg=None):
     is_(bool(a), False, msg=msg)
 
 
-def is_none(a, msg=None):
-    is_(a, None, msg=msg)
-
-
-def is_not_none(a, msg=None):
-    is_not(a, None, msg=msg)
-
-
 def is_(a, b, msg=None):
     """Assert a is b, with repr messaging on failure."""
     assert a is b, msg or "%r is not %r" % (a, b)
diff --git a/lib/sqlalchemy/testing/engines.py b/lib/sqlalchemy/testing/engines.py
index 8b334fde2..a313c298a 100644
--- a/lib/sqlalchemy/testing/engines.py
+++ b/lib/sqlalchemy/testing/engines.py
@@ -14,6 +14,7 @@ import weakref
 
 from . import config
 from .util import decorator
+from .util import gc_collect
 from .. import event
 from .. import pool
 
@@ -124,6 +125,19 @@ class ConnectionKiller(object):
         self._drop_testing_engines("function")
         self._drop_testing_engines("class")
 
+    def stop_test_class_outside_fixtures(self):
+        # ensure no refs to checked out connections at all.
+
+        if pool.base._strong_ref_connection_records:
+            gc_collect()
+
+            if pool.base._strong_ref_connection_records:
+                ln = len(pool.base._strong_ref_connection_records)
+                pool.base._strong_ref_connection_records.clear()
+                assert (
+                    False
+                ), "%d connection recs not cleared after test suite" % (ln)
+
     def final_cleanup(self):
         self.checkin_all()
         for scope in self.testing_engines:
diff --git a/lib/sqlalchemy/testing/plugin/plugin_base.py b/lib/sqlalchemy/testing/plugin/plugin_base.py
index 7851fbb3e..858814f91 100644
--- a/lib/sqlalchemy/testing/plugin/plugin_base.py
+++ b/lib/sqlalchemy/testing/plugin/plugin_base.py
@@ -586,6 +586,7 @@ def stop_test_class(cls):
 
 
 def stop_test_class_outside_fixtures(cls):
+    engines.testing_reaper.stop_test_class_outside_fixtures()
     provision.stop_test_class_outside_fixtures(config, config.db, cls)
     try:
         if not options.low_connections: