Brrr.

Weak sets have have pragmatic gotchas, explained in the comments
before the new WeakSet.as_weakref_list() method.  In essence, we
just took all the weak sets of connection objects and changed
everything so that a list of live objects is never materialized
anymore.  Also added new map()-like methods so that clients don't
usually need to be aware of the weakrefs under the covers.

trunk/src/ZODB/DB.py

@@ -130,10 +130,12 @@ class _ConnectionPool(object):
             assert result in self.all
         return result
 
-    # Return a list of all connections we currently know about.
-    def all_as_list(self):
-        return self.all.as_list()
-
+    # For every live connection c, invoke f(c).
+    def map(self, f):
+        for wr in self.all.as_weakref_list():
+            c = wr()
+            if c is not None:
+                f(c)
 
 class DB(object):
     """The Object Database
@@ -294,8 +296,7 @@ class DB(object):
         self._a()
         try:
             for pool in self._pools.values():
-                for c in pool.all_as_list():
-                    f(c)
+                pool.map(f)
         finally:
             self._r()
 
@@ -562,22 +563,26 @@ class DB(object):
     def connectionDebugInfo(self):
         result = []
         t = time()
-        for version, pool in self._pools.items():
-            for c in pool.all_as_list():
-                o = c._opened
-                d = c._debug_info
-                if d:
-                    if len(d) == 1:
-                        d = d[0]
-                else:
-                    d = ''
-                d = "%s (%s)" % (d, len(c._cache))
 
-                result.append({
-                    'opened': o and ("%s (%.2fs)" % (ctime(o), t-o)),
-                    'info': d,
-                    'version': version,
-                    })
+        def get_info(c):
+            # `result`, `time` and `version` are lexically inherited.
+            o = c._opened
+            d = c._debug_info
+            if d:
+                if len(d) == 1:
+                    d = d[0]
+            else:
+                d = ''
+            d = "%s (%s)" % (d, len(c._cache))
+
+            result.append({
+                'opened': o and ("%s (%.2fs)" % (ctime(o), t-o)),
+                'info': d,
+                'version': version,
+                })
+
+        for version, pool in self._pools.items():
+            pool.map(get_info)
         return result
 
     def getActivityMonitor(self):
@@ -614,25 +619,27 @@ class DB(object):
         # Zope will rebind this method to arbitrary user code at runtime.
         return find_global(modulename, globalname)
 
-    def setCacheSize(self, v):
+    def setCacheSize(self, size):
         self._a()
         try:
-            self._cache_size = v
+            self._cache_size = size
             pool = self._pools.get('')
             if pool is not None:
-                for c in pool.all_as_list():
-                    c._cache.cache_size = v
+                def setsize(c):
+                    c._cache.cache_size = size
+                pool.map(setsize)
         finally:
             self._r()
 
-    def setVersionCacheSize(self, v):
+    def setVersionCacheSize(self, size):
         self._a()
         try:
-            self._version_cache_size = v
+            self._version_cache_size = size
+            def setsize(c):
+                c._cache.cache_size = size
             for version, pool in self._pools.items():
                 if version:
-                    for c in pool.all_as_list():
-                        c._cache.cache_size = v
+                    pool.map(setsize)
         finally:
             self._r()
 

trunk/src/ZODB/utils.py

@@ -221,10 +221,25 @@ class WeakSet(object):
     def remove(self, obj):
         del self.data[id(obj)]
 
-    # Return a list of all the objects in the collection.
-    # Because a weak dict is used internally, iteration
-    # is dicey (the underlying dict may change size during
-    # iteration, due to gc or activity from other threads).
-    # as_list() attempts to be safe.
-    def as_list(self):
-        return self.data.values()
+    # Return a list of weakrefs to all the objects in the collection.
+    # Because a weak dict is used internally, iteration is dicey (the
+    # underlying dict may change size during iteration, due to gc or
+    # activity from other threads).  as_weakef_list() is safe.
+    #
+    # Something like this should really be a method of Python's weak dicts.
+    # If we invoke self.data.values() instead, we get back a list of live
+    # objects instead of weakrefs.  If gc occurs while this list is alive,
+    # all the objects move to an older generation (because they're strongly
+    # referenced by the list!).  They can't get collected then, until a
+    # less frequent collection of the older generation.  Before then, if we
+    # invoke self.data.values() again, they're still alive, and if gc occurs
+    # while that list is alive they're all moved to yet an older generation.
+    # And so on.  Stress tests showed that it was easy to get into a state
+    # where a WeakSet grows without bounds, despite that almost all its
+    # elements are actually trash.  By returning a list of weakrefs instead,
+    # we avoid that, although the decision to use weakrefs is now# very
+    # visible to our clients.
+    def as_weakref_list(self):
+        # We're cheating by breaking into the internals of Python's
+        # WeakValueDictionary here (accessing its .data attribute).
+        return self.data.data.values()

trunk/src/transaction/_manager.py

@@ -43,12 +43,12 @@ class TransactionManager(object):
     def begin(self):
         if self._txn is not None:
             self._txn.abort()
-        self._txn = Transaction(self._synchs.as_list(), self)
+        self._txn = Transaction(self._synchs.as_weakref_list(), self)
         return self._txn
 
     def get(self):
         if self._txn is None:
-            self._txn = Transaction(self._synchs.as_list(), self)
+            self._txn = Transaction(self._synchs.as_weakref_list(), self)
         return self._txn
 
     def free(self, txn):
@@ -82,7 +82,7 @@ class ThreadTransactionManager(object):
             txn.abort()
         synchs = self._synchs.get(tid)
         if synchs is not None:
-            synchs = synchs.as_list()
+            synchs = synchs.as_weakref_list()
         txn = self._txns[tid] = Transaction(synchs, self)
         return txn
 
@@ -92,7 +92,7 @@ class ThreadTransactionManager(object):
         if txn is None:
             synchs = self._synchs.get(tid)
             if synchs is not None:
-                synchs = synchs.as_list()
+                synchs = synchs.as_weakref_list()
             txn = self._txns[tid] = Transaction(synchs, self)
         return txn
 

trunk/src/transaction/_transaction.py

@@ -138,6 +138,7 @@ import sys
 import thread
 import warnings
 import traceback
+import weakref
 from cStringIO import StringIO
 
 # Sigh.  In the maze of __init__.py's, ZODB.__init__.py takes 'get'
@@ -203,6 +204,14 @@ class Transaction(object):
         # raised, incorporating this traceback.
         self._failure_traceback = None
 
+    # Invoke f(synch) for each synch in self._synchronizers.
+    def _synch_map(self, f):
+        for wr in self._synchronizers:
+            assert isinstance(wr, weakref.ref)
+            synch = wr()
+            if synch is not None:
+                f(synch)
+
     # Raise TransactionFailedError, due to commit()/join()/register()
     # getting called when the current transaction has already suffered
     # a commit failure.
@@ -286,8 +295,7 @@ class Transaction(object):
             self.commit(True)
 
         if not subtransaction:
-            for s in self._synchronizers:
-                s.beforeCompletion(self)
+            self._synch_map(lambda s: s.beforeCompletion(self))
             self.status = Status.COMMITTING
 
         try:
@@ -311,8 +319,7 @@ class Transaction(object):
             self.status = Status.COMMITTED
             if self._manager:
                 self._manager.free(self)
-            for s in self._synchronizers:
-                s.afterCompletion(self)
+            self._synch_map(lambda s: s.afterCompletion(self))
             self.log.debug("commit")
 
     def _commitResources(self, subtransaction):
@@ -360,8 +367,7 @@ class Transaction(object):
                 self._cleanup(L)
             finally:
                 if not subtransaction:
-                    for s in self._synchronizers:
-                        s.afterCompletion(self)
+                    self._synch_map(lambda s: s.afterCompletion(self))
             raise t, v, tb
 
     def _cleanup(self, L):
@@ -427,8 +433,7 @@ class Transaction(object):
 
     def abort(self, subtransaction=False):
         if not subtransaction:
-            for s in self._synchronizers:
-                s.beforeCompletion(self)
+            self._synch_map(lambda s: s.beforeCompletion(self))
 
         if subtransaction and self._nonsub:
             from ZODB.POSException import TransactionError
@@ -458,8 +463,7 @@ class Transaction(object):
         if not subtransaction:
             if self._manager:
                 self._manager.free(self)
-            for s in self._synchronizers:
-                s.afterCompletion(self)
+            self._synch_map(lambda s: s.afterCompletion(self))
             self.log.debug("abort")
 
         if tb is not None: