Replace reset_agent with direct call from connection

Fixed a regression where the "reset agent" of the connection pool wasn't
really being utilized by the :class:`_engine.Connection` when it were
closed, and also leading to a double-rollback scenario that was somewhat
wasteful.   The newer architecture of the engine has been updated so that
the connection pool "reset-on-return" logic will be skipped when the
:class:`_engine.Connection` explicitly closes out the transaction before
returning the pool to the connection.

Fixes: #6004
Change-Id: I5d2ac16cac71aa45a00b4b7481d7268bd828a168

2 files changed, 40 insertions, 96 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index b5a37f67e..362c811ee 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -1067,20 +1067,19 @@ class Connection(Connectable):
 
         if self._transaction:
             self._transaction.close()
+            skip_reset = True
+        else:
+            skip_reset = False
 
         if self._dbapi_connection is not None:
             conn = self._dbapi_connection
 
-            # this will do a reset-on-return every time, even if we
-            # called rollback() already. it might be worth optimizing
-            # this for the case that we are able to close without issue
-            conn.close()
-
-            # this is in fact never true outside of a bunch of
-            # artificial scenarios created by the test suite and its
-            # fixtures.  the reset_agent should no longer be necessary.
-            if conn._reset_agent is self._transaction:
-                conn._reset_agent = None
+            # as we just closed the transaction, close the connection
+            # pool connection without doing an additional reset
+            if skip_reset:
+                conn._close_no_reset()
+            else:
+                conn.close()
 
             # There is a slight chance that conn.close() may have
             # triggered an invalidation here in which case
@@ -2309,36 +2308,14 @@ class RootTransaction(Transaction):
 
         self.is_active = True
 
-        # the SingletonThreadPool used with sqlite memory can share the same
-        # DBAPI connection / fairy among multiple Connection objects.  while
-        # this is not ideal, it is a still-supported use case which at the
-        # moment occurs in the test suite due to how some of pytest fixtures
-        # work out
-        if connection._dbapi_connection._reset_agent is None:
-            connection._dbapi_connection._reset_agent = self
-
     def _deactivate_from_connection(self):
         if self.is_active:
             assert self.connection._transaction is self
             self.is_active = False
 
-            if (
-                self.connection._dbapi_connection is not None
-                and self.connection._dbapi_connection._reset_agent is self
-            ):
-                self.connection._dbapi_connection._reset_agent = None
-
         elif self.connection._transaction is not self:
             util.warn("transaction already deassociated from connection")
 
-        # we have tests that want to make sure the pool handles this
-        # correctly.  TODO: how to disable internal assertions cleanly?
-        # else:
-        #    if self.connection._dbapi_connection is not None:
-        #        assert (
-        #            self.connection._dbapi_connection._reset_agent is not self
-        #        )
-
     def _do_deactivate(self):
         # called from a MarkerTransaction to cancel this root transaction.
         # the transaction stays in place as connection._transaction, but
diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index 9b4e61fc3..6ec489604 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -106,7 +106,9 @@ class Pool(log.Identified):
              logging.
 
         :param reset_on_return: Determine steps to take on
-          connections as they are returned to the pool.
+          connections as they are returned to the pool, which were
+          not otherwise handled by a :class:`_engine.Connection`.
+
           reset_on_return can have any of these values:
 
           * ``"rollback"`` - call rollback() on the connection,
@@ -124,22 +126,19 @@ class Pool(log.Identified):
             any data changes present on the transaction
             are committed unconditionally.
           * ``None`` - don't do anything on the connection.
-            This setting should generally only be made on a database
-            that has no transaction support at all,
-            namely MySQL MyISAM; when used on this backend, performance
-            can be improved as the "rollback" call is still expensive on
-            MySQL.   It is **strongly recommended** that this setting not be
-            used for transaction-supporting databases in conjunction with
-            a persistent pool such as :class:`.QueuePool`, as it opens
-            the possibility for connections still in a transaction to be
-            idle in the pool.   The setting may be appropriate in the
-            case of :class:`.NullPool` or special circumstances where
-            the connection pool in use is not being used to maintain connection
-            lifecycle.
+            This setting is only appropriate if the database / DBAPI
+            works in pure "autocommit" mode at all times, or if the
+            application uses the :class:`_engine.Engine` with consistent
+            connectivity patterns.   See the section
+            :ref:`pool_reset_on_return` for more details.
 
           * ``False`` - same as None, this is here for
             backwards compatibility.
 
+         .. seealso::
+
+            :ref:`pool_reset_on_return`
+
         :param events: a list of 2-tuples, each of the form
          ``(callable, target)`` which will be passed to :func:`.event.listen`
          upon construction.   Provided here so that event listeners
@@ -429,7 +428,7 @@ class _ConnectionRecord(object):
         rec.fairy_ref = ref = weakref.ref(
             fairy,
             lambda ref: _finalize_fairy
-            and _finalize_fairy(None, rec, pool, ref, echo),
+            and _finalize_fairy(None, rec, pool, ref, echo, True),
         )
         _strong_ref_connection_records[ref] = rec
         if echo:
@@ -612,6 +611,7 @@ def _finalize_fairy(
     pool,
     ref,  # this is None when called directly, not by the gc
     echo,
+    reset=True,
     fairy=None,
 ):
     """Cleanup for a :class:`._ConnectionFairy` whether or not it's already
@@ -647,7 +647,11 @@ def _finalize_fairy(
     if connection is not None:
         if connection_record and echo:
             pool.logger.debug(
-                "Connection %r being returned to pool", connection
+                "Connection %r being returned to pool%s",
+                connection,
+                ", transaction state was already reset by caller"
+                if not reset
+                else "",
             )
 
         try:
@@ -655,7 +659,7 @@ def _finalize_fairy(
                 connection, connection_record, echo
             )
             assert fairy.connection is connection
-            if can_manipulate_connection:
+            if reset and can_manipulate_connection:
                 fairy._reset(pool)
 
             if detach:
@@ -738,24 +742,6 @@ class _ConnectionFairy(object):
 
     """
 
-    _reset_agent = None
-    """Refer to an object with a ``.commit()`` and ``.rollback()`` method;
-    if non-None, the "reset-on-return" feature will call upon this object
-    rather than directly against the dialect-level do_rollback() and
-    do_commit() methods.
-
-    In practice, a :class:`_engine.Connection` assigns a :class:`.Transaction`
-    object
-    to this variable when one is in scope so that the :class:`.Transaction`
-    takes the job of committing or rolling back on return if
-    :meth:`_engine.Connection.close` is called while the :class:`.Transaction`
-    still exists.
-
-    This is essentially an "event handler" of sorts but is simplified as an
-    instance variable both for performance/simplicity as well as that there
-    can only be one "reset agent" at a time.
-    """
-
     @classmethod
     def _checkout(cls, pool, threadconns=None, fairy=None):
         if not fairy:
@@ -856,13 +842,14 @@ class _ConnectionFairy(object):
     def _checkout_existing(self):
         return _ConnectionFairy._checkout(self._pool, fairy=self)
 
-    def _checkin(self):
+    def _checkin(self, reset=True):
         _finalize_fairy(
             self.connection,
             self._connection_record,
             self._pool,
             None,
             self._echo,
+            reset=reset,
             fairy=self,
         )
         self.connection = None
@@ -876,41 +863,16 @@ class _ConnectionFairy(object):
         if pool._reset_on_return is reset_rollback:
             if self._echo:
                 pool.logger.debug(
-                    "Connection %s rollback-on-return%s",
-                    self.connection,
-                    ", via agent" if self._reset_agent else "",
+                    "Connection %s rollback-on-return", self.connection
                 )
-            if self._reset_agent:
-                if not self._reset_agent.is_active:
-                    util.warn(
-                        "Reset agent is not active.  "
-                        "This should not occur unless there was already "
-                        "a connectivity error in progress."
-                    )
-                    pool._dialect.do_rollback(self)
-                else:
-                    self._reset_agent.rollback()
-            else:
-                pool._dialect.do_rollback(self)
+            pool._dialect.do_rollback(self)
         elif pool._reset_on_return is reset_commit:
             if self._echo:
                 pool.logger.debug(
-                    "Connection %s commit-on-return%s",
+                    "Connection %s commit-on-return",
                     self.connection,
-                    ", via agent" if self._reset_agent else "",
                 )
-            if self._reset_agent:
-                if not self._reset_agent.is_active:
-                    util.warn(
-                        "Reset agent is not active.  "
-                        "This should not occur unless there was already "
-                        "a connectivity error in progress."
-                    )
-                    pool._dialect.do_commit(self)
-                else:
-                    self._reset_agent.commit()
-            else:
-                pool._dialect.do_commit(self)
+            pool._dialect.do_commit(self)
 
     @property
     def _logger(self):
@@ -1032,3 +994,8 @@ class _ConnectionFairy(object):
         self._counter -= 1
         if self._counter == 0:
             self._checkin()
+
+    def _close_no_reset(self):
+        self._counter -= 1
+        if self._counter == 0:
+            self._checkin(reset=False)