Merge pull request #146 from zopefoundation/storage-afterCompletion2

Notify storages that transactions have completed.

CHANGES.rst

@@ -2,14 +2,23 @@
  Change History
 ================
 
-5.1.2 (unreleased)
+5.2.0 (unreleased)
 ==================
 
+- Call new afterCompletion API on storages to allow them to free
+  resources after transaction complete.  See:
+  https://github.com/zodb/relstorage/issues/147
+
+- Take advantage of the new transaction-manager explicit mode to avoid
+  starting transactions unnecessarily when transactions end.
+
 - ``Connection.new_oid`` delegates to its storage, not the DB. This is
   helpful for improving concurrency in MVCC storages like RelStorage.
   See `issue 139 <https://github.com/zopefoundation/ZODB/issues/139`_.
+
 - ``persistent`` is no longer required at setup time.
   See `issue 119 <https://github.com/zopefoundation/ZODB/issues/119>`_.
+
 - ``Connection.close`` and ``Connection.open`` no longer race on
   ``self.transaction_manager``, which could lead to
   ``AttributeError``. This was a bug introduced in 5.0.1. See `issue

src/ZODB/Connection.py

@@ -312,6 +312,9 @@ class Connection(ExportImport, object):
         # Drop transaction manager to release resources and help prevent errors
         self.transaction_manager = None
 
+        if hasattr(self._storage, 'afterCompletion'):
+            self._storage.afterCompletion()
+
         if primary:
             for connection in self.connections.values():
                 if connection is not self:
@@ -406,7 +409,6 @@ class Connection(ExportImport, object):
 
     def abort(self, transaction):
         """Abort a transaction and forget all changes."""
-
         # The order is important here.  We want to abort registered
         # objects before we process the cache.  Otherwise, we may un-add
         # objects added in savepoints.  If they've been modified since
@@ -480,7 +482,6 @@ class Connection(ExportImport, object):
 
     def commit(self, transaction):
         """Commit changes to an object"""
-
         transaction = transaction.data(self)
 
         if self._savepoint_storage is not None:
@@ -745,9 +746,6 @@ class Connection(ExportImport, object):
         except AttributeError:
             assert self._storage is None
 
-        # Now is a good time to collect some garbage.
-        self._cache.incrgc()
-
     def afterCompletion(self, transaction):
         # Note that we we call newTransaction here for 2 reasons:
         # a) Applying invalidations early frees up resources
@@ -757,8 +755,15 @@ class Connection(ExportImport, object):
         #    finalizing previous ones without calling begin.  We pass
         #    False to avoid possiblyt expensive sync calls to not
         #    penalize well-behaved applications that call begin.
+        if hasattr(self._storage, 'afterCompletion'):
+            self._storage.afterCompletion()
+
+        if not self.explicit_transactions:
             self.newTransaction(transaction, False)
 
+        # Now is a good time to collect some garbage.
+        self._cache.incrgc()
+
     # Transaction-manager synchronization -- ISynchronizer
     ##########################################################################
 
@@ -772,8 +777,9 @@ class Connection(ExportImport, object):
         return self._reader.getState(p)
 
     def setstate(self, obj):
-        """Turns the ghost 'obj' into a real object by loading its state from
-        the database."""
+        """Load the state for an (ghost) object
+        """
+
         oid = obj._p_oid
 
         if self.opened is None:
@@ -887,12 +893,16 @@ class Connection(ExportImport, object):
 
         self.transaction_manager = transaction_manager
 
+        self.explicit_transactions = getattr(transaction_manager,
+                                             'explicit', False)
+
         self.opened = time.time()
 
         if self._reset_counter != global_reset_counter:
             # New code is in place.  Start a new cache.
             self._resetCache()
 
+        if not self.explicit_transactions:
             # This newTransaction is to deal with some pathalogical cases:
             #
             # a) Someone opens a connection when a transaction isn't
@@ -906,8 +916,10 @@ class Connection(ExportImport, object):
             # b) Lots of tests assume that connection transaction
             #    information is set on open.
             #
-        # Fortunately, this is a cheap operation.  It doesn't really
-        # cost much, if anything.
+            # Fortunately, this is a cheap operation.  It doesn't
+            # really cost much, if anything.  Well, except for
+            # RelStorage, in which case it adds a server round
+            # trip.
             self.newTransaction(None, False)
 
         transaction_manager.registerSynch(self)

src/ZODB/interfaces.py

@@ -1243,6 +1243,16 @@ class IMVCCPrefetchStorage(IMVCCStorage):
         more than once.
         """
 
+class IMVCCAfterCompletionStorage(IMVCCStorage):
+
+    def afterCompletion():
+        """Notify a storage that a transaction has ended.
+
+        The storage may choose to use this opportunity to release resources.
+
+        See ``transaction.interfaces.ISynchronizer.afterCompletion``.
+        """
+
 class IStorageCurrentRecordIteration(IStorage):
 
     def record_iternext(next=None):

src/ZODB/tests/testConnection.py

@@ -1060,6 +1060,7 @@ def doctest_lp485456_setattr_in_setstate_doesnt_cause_multiple_stores():
     >>> conn.close()
     """
 
+
 class _PlayPersistent(Persistent):
     def setValueWithSize(self, size=0): self.value = size*' '
     __init__ = setValueWithSize
@@ -1301,14 +1302,76 @@ class StubStorage:
         return z64
 
 
-class TestConnectionInterface(unittest.TestCase):
+class TestConnection(unittest.TestCase):
 
     def test_connection_interface(self):
         from ZODB.interfaces import IConnection
         db = databaseFromString("<zodb>\n<mappingstorage/>\n</zodb>")
         cn = db.open()
         verifyObject(IConnection, cn)
+        db.close()
+
+    def test_storage_afterCompletionCalled(self):
+        db = ZODB.DB(None)
+        conn = db.open()
+        data = []
+        conn._storage.afterCompletion = lambda : data.append(None)
+        conn.transaction_manager.commit()
+        self.assertEqual(len(data), 1)
+        conn.close()
+        self.assertEqual(len(data), 2)
+        db.close()
+
+    def test_explicit_transactions_no_newTransactuon_on_afterCompletion(self):
+        syncs = []
+        from .MVCCMappingStorage import MVCCMappingStorage
+        storage = MVCCMappingStorage()
+
+        new_instance = storage.new_instance
+        def new_instance2():
+            inst = new_instance()
+            sync = inst.sync
+            def sync2(*args):
+                sync()
+                syncs.append(1)
+            inst.sync = sync2
+            return inst
+
+        storage.new_instance = new_instance2
+
+        db = ZODB.DB(storage)
+        del syncs[:] # Need to do this to clear effect of getting the
+                     # root object
+
+        # We don't want to depend on latest transaction package, so
+        # just set attr for test:
+        tm = transaction.TransactionManager()
+        tm.explicit = True
+
+        conn = db.open(tm)
+        self.assertEqual(len(syncs), 0)
+        conn.transaction_manager.begin()
+        self.assertEqual(len(syncs), 1)
+        conn.transaction_manager.commit()
+        self.assertEqual(len(syncs), 1)
+        conn.transaction_manager.begin()
+        self.assertEqual(len(syncs), 2)
+        conn.transaction_manager.abort()
+        self.assertEqual(len(syncs), 2)
+        conn.close()
+        self.assertEqual(len(syncs), 2)
+
+        # For reference, in non-explicit mode:
+        conn = db.open()
+        self.assertEqual(len(syncs), 3)
+        conn._storage.sync = syncs.append
+        conn.transaction_manager.begin()
+        self.assertEqual(len(syncs), 4)
+        conn.transaction_manager.abort()
+        self.assertEqual(len(syncs), 5)
+        conn.close()
 
+        db.close()
 
 class StubDatabase:
 
@@ -1330,6 +1393,6 @@ def test_suite():
     s = unittest.makeSuite(ConnectionDotAdd)
     s.addTest(unittest.makeSuite(SetstateErrorLoggingTests))
     s.addTest(doctest.DocTestSuite(checker=checker))
-    s.addTest(unittest.makeSuite(TestConnectionInterface))
+    s.addTest(unittest.makeSuite(TestConnection))
     s.addTest(unittest.makeSuite(EstimatedSizeTests))
     return s