[Zodb-checkins] SVN: ZODB/branches/3.8/src/ZEO/ Made caches thread safe. In theory, caches are protected by ZEO

Jim Fulton jim at zope.com
Tue Jul 8 19:39:36 EDT 2008


Log message for revision 88128:
  Made caches thread safe.  In theory, caches are protected by ZEO
  clients, but ZEO clients haven't provided very good protection,
  leading to cache corruption.  We'll hopefully fix these client issues,
  which cause other problems beside cache corruption, but it seems
  prudent to provide low-level cache protection.
  

Changed:
  U   ZODB/branches/3.8/src/ZEO/ClientStorage.py
  U   ZODB/branches/3.8/src/ZEO/cache.py
  U   ZODB/branches/3.8/src/ZEO/tests/test_cache.py

-=-
Modified: ZODB/branches/3.8/src/ZEO/ClientStorage.py
===================================================================
--- ZODB/branches/3.8/src/ZEO/ClientStorage.py	2008-07-08 22:07:13 UTC (rev 88127)
+++ ZODB/branches/3.8/src/ZEO/ClientStorage.py	2008-07-08 23:39:35 UTC (rev 88128)
@@ -338,8 +338,6 @@
         else:
             cache_path = None
         self._cache = self.ClientCacheClass(cache_path, size=cache_size)
-        # TODO:  maybe there's a better time to open the cache?  Unclear.
-        self._cache.open()
 
         self._rpc_mgr = self.ConnectionManagerClass(addr, self,
                                                     tmin=min_disconnect_poll,

Modified: ZODB/branches/3.8/src/ZEO/cache.py
===================================================================
--- ZODB/branches/3.8/src/ZEO/cache.py	2008-07-08 22:07:13 UTC (rev 88127)
+++ ZODB/branches/3.8/src/ZEO/cache.py	2008-07-08 23:39:35 UTC (rev 88128)
@@ -30,6 +30,7 @@
 import logging
 import os
 import tempfile
+import threading
 import time
 
 import ZODB.fsIndex
@@ -121,7 +122,22 @@
 # to the end of the file that the new object can't fit in one
 # contiguous chunk, currentofs is reset to ZEC3_HEADER_SIZE first.
 
+class locked(object):
 
+    def __init__(self, func):
+        self.func = func
+
+    def __get__(self, inst, class_):
+        if inst is None:
+            return self
+        def call(*args, **kw):
+            inst._lock.acquire()
+            try:
+                return self.func(inst, *args, **kw)
+            finally:
+                inst._lock.release()
+        return call
+
 class ClientCache(object):
     """A simple in-memory cache."""
 
@@ -200,6 +216,10 @@
 
         self._setup_trace(path)
 
+        self.open()
+
+        self._lock = threading.RLock()
+
     # Backward compatibility. Client code used to have to use the fc
     # attr to get to the file cache to get cache stats.
     @property
@@ -353,6 +373,7 @@
     # instance, and also written out near the start of the cache file.  The
     # new tid must be strictly greater than our current idea of the most
     # recent tid.
+    @locked
     def setLastTid(self, tid):
         if self.tid is not None and tid <= self.tid:
             raise ValueError("new last tid (%s) must be greater than "
@@ -369,10 +390,11 @@
     # @return a transaction id
     # @defreturn string, or None if no transaction is yet known
     def getLastTid(self):
-        if self.tid == z64:
+        tid = self.tid
+        if tid == z64:
             return None
         else:
-            return self.tid
+            return tid
 
     ##
     # Return the current data record for oid and version.
@@ -382,6 +404,7 @@
     #         in the cache
     # @defreturn 3-tuple: (string, string, string)
 
+    @locked
     def load(self, oid, version=""):
         ofs = self.current.get(oid)
         if ofs is None:
@@ -414,6 +437,7 @@
     # @return data record, serial number, start tid, and end tid
     # @defreturn 4-tuple: (string, string, string, string)
 
+    @locked
     def loadBefore(self, oid, before_tid):
         noncurrent_for_oid = self.noncurrent.get(u64(oid))
         if noncurrent_for_oid is None:
@@ -455,6 +479,7 @@
     # @defreturn string or None
 
     # XXX This approac is wrong, but who cares
+    @locked
     def modifiedInVersion(self, oid):
         ofs = self.current.get(oid)
         if ofs is None:
@@ -482,6 +507,7 @@
     # @param data the actual data
     # @exception ValueError tried to store non-current version data
 
+    @locked
     def store(self, oid, version, start_tid, end_tid, data):
         # It's hard for the client to avoid storing the same object
         # more than once.  One case is when the client requests
@@ -586,6 +612,7 @@
     # @param tid the id of the transaction that wrote a new revision of oid,
     #        or None to forget all cached info about oid (version, current
     #        revision, and non-current revisions)
+    @locked
     def invalidate(self, oid, version, tid):
         if tid > self.tid and tid is not None:
             self.setLastTid(tid)
@@ -630,17 +657,25 @@
         seek = self.f.seek
         read = self.f.read
         for oid, ofs in self.current.iteritems():
-            seek(ofs)
-            assert read(1) == 'a', (ofs, self.f.tell(), oid)
-            size, saved_oid, tid, end_tid, lver = unpack(">I8s8s8sh", read(30))
-            assert saved_oid == oid, (ofs, self.f.tell(), oid, saved_oid)
-            assert end_tid == z64, (ofs, self.f.tell(), oid)
-            if lver:
-                version = read(lver)
-            else:
-                version = ''
-            yield oid, tid, version
 
+            self._lock.acquire()
+            try:
+                seek(ofs)
+                assert read(1) == 'a', (ofs, self.f.tell(), oid)
+                size, saved_oid, tid, end_tid, lver = unpack(
+                    ">I8s8s8sh", read(30))
+                assert saved_oid == oid, (ofs, self.f.tell(), oid, saved_oid)
+                assert end_tid == z64, (ofs, self.f.tell(), oid)
+                if lver:
+                    version = read(lver)
+                else:
+                    version = ''
+                result = oid, tid, version
+            finally:
+                self._lock.release()
+
+            yield result
+
     def dump(self):
         from ZODB.utils import oid_repr
         print "cache size", len(self)

Modified: ZODB/branches/3.8/src/ZEO/tests/test_cache.py
===================================================================
--- ZODB/branches/3.8/src/ZEO/tests/test_cache.py	2008-07-08 22:07:13 UTC (rev 88127)
+++ ZODB/branches/3.8/src/ZEO/tests/test_cache.py	2008-07-08 23:39:35 UTC (rev 88128)
@@ -35,7 +35,6 @@
         # testSerialization reads the entire file into a string, it's not
         # good to leave it that big.
         self.cache = ZEO.cache.ClientCache(size=1024**2)
-        self.cache.open()
 
     def tearDown(self):
         if self.cache.path:
@@ -142,7 +141,6 @@
         dst.write(src.read(self.cache.maxsize))
         dst.close()
         copy = ZEO.cache.ClientCache(path)
-        copy.open()
 
         # Verify that internals of both objects are the same.
         # Could also test that external API produces the same results.
@@ -158,7 +156,6 @@
         if self.cache.path:
             os.remove(self.cache.path)
         cache = ZEO.cache.ClientCache(size=50)
-        cache.open()
 
         # We store an object that is a bit larger than the cache can handle.
         cache.store(n1, '', n2, None, "x"*64)
@@ -174,7 +171,6 @@
         if self.cache.path:
             os.remove(self.cache.path)
         cache = ZEO.cache.ClientCache(size=50)
-        cache.open()
 
         # We store an object that is a bit larger than the cache can handle.
         cache.store(n1, '', n2, n3, "x"*64)
@@ -218,7 +214,6 @@
     ...     _ = os.spawnl(os.P_WAIT, sys.executable, sys.executable, 't')
     ...     if os.path.exists('cache'):
     ...         cache = ZEO.cache.ClientCache('cache')
-    ...         cache.open()
     ...         cache.close()
     ...         os.remove('cache')
     ...         os.remove('cache.lock')
@@ -238,7 +233,6 @@
     >>> cache.store(ZODB.utils.p64(1), '', ZODB.utils.p64(1), None, data)
     >>> cache.close()
     >>> cache = ZEO.cache.ClientCache('cache', 1000)
-    >>> cache.open()
     >>> cache.store(ZODB.utils.p64(2), '', ZODB.utils.p64(2), None, 'XXX')
 
     >>> cache.close()
@@ -255,6 +249,57 @@
 
     >>> cache.close()
     """,
+
+    thread_safe =
+    r"""
+
+    >>> import ZEO.cache, ZODB.utils
+    >>> cache = ZEO.cache.ClientCache('cache', 1000000)
+
+    >>> for i in range(100):
+    ...     cache.store(ZODB.utils.p64(i), '', ZODB.utils.p64(1), None, '0')
+
+    >>> import random, sys, threading
+    >>> random = random.Random(0)
+    >>> stop = False
+    >>> read_failure = None
+
+    >>> def read_thread():
+    ...     def pick_oid():
+    ...         return ZODB.utils.p64(random.randint(0,99))
+    ...
+    ...     try:
+    ...         while not stop:
+    ...             cache.load(pick_oid())
+    ...             cache.loadBefore(pick_oid(), ZODB.utils.p64(2))
+    ...             cache.modifiedInVersion(pick_oid())
+    ...     except:
+    ...         global read_failure
+    ...         read_failure = sys.exc_info()
+
+    >>> thread = threading.Thread(target=read_thread)
+    >>> thread.start()
+
+    >>> for tid in range(2,10):
+    ...     for oid in range(100):
+    ...         oid = ZODB.utils.p64(oid)
+    ...         cache.invalidate(oid, '', ZODB.utils.p64(tid))
+    ...         cache.store(oid, '', ZODB.utils.p64(tid), None, str(tid))
+
+    >>> stop = True
+    >>> thread.join()
+    >>> if read_failure:
+    ...    print 'Read failure:'
+    ...    import traceback
+    ...    traceback.print_exception(*read_failure)
+
+    >>> expected = '9', ZODB.utils.p64(9), ''
+    >>> for oid in range(100):
+    ...     loaded = cache.load(ZODB.utils.p64(oid))
+    ...     if loaded != expected:
+    ...         print oid, loaded
+    
+    """,
     )
 
 



More information about the Zodb-checkins mailing list