[Zodb-checkins] CVS: ZODB3/ZODB - FileStorage.py:1.139.2.1 DemoStorage.py:1.21.2.1 DB.py:1.55.2.1 Connection.py:1.100.2.1 BaseStorage.py:1.36.2.1

Jeremy Hylton jeremy at zope.com
Tue Oct 7 01:11:04 EDT 2003


Update of /cvs-repository/ZODB3/ZODB
In directory cvs.zope.org:/tmp/cvs-serv28710/ZODB

Modified Files:
      Tag: ZODB3-mvcc-2-branch
	FileStorage.py DemoStorage.py DB.py Connection.py 
	BaseStorage.py 
Log Message:
Merge changes from the ZODB3-mvcc-branch.

This new branch is relative to the head, instead of the
Zope-2_7-branch.


=== ZODB3/ZODB/FileStorage.py 1.139 => 1.139.2.1 ===
--- ZODB3/ZODB/FileStorage.py:1.139	Thu Oct  2 18:14:04 2003
+++ ZODB3/ZODB/FileStorage.py	Tue Oct  7 01:10:31 2003
@@ -643,7 +643,7 @@
             spos = h[-8:]
             srcpos = u64(spos)
         self._toid2serial_delete.update(current_oids)
-        return oids
+        return self._serial, oids
 
     def getSize(self):
         return self._pos
@@ -1315,7 +1315,7 @@
         # It's too painful to try to update them to correct current
         # values instead.
         self._toid2serial_delete.update(tindex)
-        return tindex.keys()
+        return self._serial, tindex.keys()
 
     def _txn_find(self, tid, stop_at_pack):
         pos = self._pos
@@ -1460,9 +1460,6 @@
         self._lock_acquire()
         try:
             r=[]
-            file=self._file
-            seek=file.seek
-            read=file.read
             try:
                 pos=self._index[oid]
             except KeyError:
@@ -1473,14 +1470,14 @@
 
             while 1:
                 if len(r) >= size: return r
-                seek(pos)
-                h=read(DATA_HDR_LEN)
+                self._file.seek(pos)
+                h=self._file.read(DATA_HDR_LEN)
                 doid,serial,prev,tloc,vlen,plen = unpack(DATA_HDR, h)
                 prev=u64(prev)
 
                 if vlen:
-                    read(16)
-                    version=read(vlen)
+                    self._file.read(16)
+                    version = self._file.read(vlen)
                     if wantver is not None and version != wantver:
                         if prev:
                             pos=prev
@@ -1491,13 +1488,15 @@
                     version=''
                     wantver=None
 
-                seek(u64(tloc))
-                h=read(TRANS_HDR_LEN)
+                self._file.seek(u64(tloc))
+                h = self._file.read(TRANS_HDR_LEN)
                 tid, stl, status, ul, dl, el = unpack(TRANS_HDR,h)
-                user_name=read(ul)
-                description=read(dl)
-                if el: d=loads(read(el))
-                else: d={}
+                user_name = self._file.read(ul)
+                description = self._file.read(dl)
+                if el:
+                    d=loads(self._file.read(el))
+                else:
+                    d={}
 
                 d['time']=TimeStamp(serial).timeTime()
                 d['user_name']=user_name


=== ZODB3/ZODB/DemoStorage.py 1.21 => 1.21.2.1 ===
--- ZODB3/ZODB/DemoStorage.py:1.21	Thu Oct  2 14:17:19 2003
+++ ZODB3/ZODB/DemoStorage.py	Tue Oct  7 01:10:31 2003
@@ -148,7 +148,7 @@
                     # effectively, delete the thing
                     self._tindex.append([oid, None, r, None, None])
 
-            return oids
+            return self._serial, oids
 
         finally: self._lock_release()
 
@@ -181,7 +181,7 @@
                 tindex.append([oid, newserial, r, new_vdata, p])
 
 
-            return oids
+            return self._serial, oids
 
         finally: self._lock_release()
 


=== ZODB3/ZODB/DB.py 1.55 => 1.55.2.1 ===
--- ZODB3/ZODB/DB.py:1.55	Thu Oct  2 14:17:19 2003
+++ ZODB3/ZODB/DB.py	Tue Oct  7 01:10:32 2003
@@ -74,7 +74,7 @@
         self._version_cache_size=version_cache_size
         self._version_cache_deactivate_after = version_cache_deactivate_after
 
-        self._miv_cache={}
+        self._miv_cache = {}
 
         # Setup storage
         self._storage=storage
@@ -303,8 +303,7 @@
     def importFile(self, file):
         raise 'Not yet implemented'
 
-    def invalidate(self, oids, connection=None, version='',
-                   rc=sys.getrefcount):
+    def invalidate(self, tid, oids, connection=None, version=''):
         """Invalidate references to a given oid.
 
         This is used to indicate that one of the connections has committed a
@@ -326,21 +325,21 @@
             for cc in allocated:
                 if (cc is not connection and
                     (not version or cc._version==version)):
-                    if rc(cc) <= 3:
+                    if sys.getrefcount(cc) <= 3:
                         cc.close()
-                    cc.invalidate(oids)
+                    cc.invalidate(tid, oids)
 
-        temps=self._temps
-        if temps:
+        if self._temps:
             t=[]
-            for cc in temps:
-                if rc(cc) > 3:
+            for cc in self._temps:
+                if sys.getrefcount(cc) > 3:
                     if (cc is not connection and
-                        (not version or cc._version==version)):
-                        cc.invalidate(oids)
+                        (not version or cc._version == version)):
+                        cc.invalidate(tid, oids)
                     t.append(cc)
-                else: cc.close()
-            self._temps=t
+                else:
+                    cc.close()
+            self._temps = t
 
     def modifiedInVersion(self, oid):
         h=hash(oid)%131
@@ -356,7 +355,7 @@
         return len(self._storage)
 
     def open(self, version='', transaction=None, temporary=0, force=None,
-             waitflag=1):
+             waitflag=1, mvcc=True):
         """Return a object space (AKA connection) to work in
 
         The optional version argument can be used to specify that a
@@ -374,25 +373,25 @@
         try:
 
             if transaction is not None:
-                connections=transaction._connections
+                connections = transaction._connections
                 if connections:
                     if connections.has_key(version) and not temporary:
                         return connections[version]
                 else:
-                    transaction._connections=connections={}
-                transaction=transaction._connections
-
+                    transaction._connections = connections = {}
+                transaction = transaction._connections
 
             if temporary:
                 # This is a temporary connection.
                 # We won't bother with the pools.  This will be
                 # a one-use connection.
-                c=self.klass(
-                    version=version,
-                    cache_size=self._version_cache_size)
+                c = self.klass(version=version,
+                               cache_size=self._version_cache_size,
+                               mvcc=mvcc)
                 c._setDB(self)
                 self._temps.append(c)
-                if transaction is not None: transaction[id(c)]=c
+                if transaction is not None:
+                    transaction[id(c)] = c
                 return c
 
 
@@ -433,18 +432,18 @@
 
 
             if not pool:
-                c=None
+                c = None
                 if version:
                     if self._version_pool_size > len(allocated) or force:
-                        c=self.klass(
-                            version=version,
-                            cache_size=self._version_cache_size)
+                        c = self.klass(version=version,
+                                       cache_size=self._version_cache_size,
+                                       mvcc=mvcc)
                         allocated.append(c)
                         pool.append(c)
                 elif self._pool_size > len(allocated) or force:
-                    c=self.klass(
-                        version=version,
-                        cache_size=self._cache_size)
+                    c = self.klass(version=version,
+                                   cache_size=self._cache_size,
+                                   mvcc=mvcc)
                     allocated.append(c)
                     pool.append(c)
 
@@ -459,7 +458,7 @@
                             pool_lock.release()
                     else: return
 
-            elif len(pool)==1:
+            elif len(pool) == 1:
                 # Taking last one, lock the pool
                 # Note that another thread might grab the lock
                 # before us, so we might actually block, however,
@@ -473,14 +472,15 @@
                     # but it could be higher due to a race condition.
                     pool_lock.release()
 
-            c=pool[-1]
+            c = pool[-1]
             del pool[-1]
             c._setDB(self)
             for pool, allocated in pooll:
                 for cc in pool:
                     cc._incrgc()
 
-            if transaction is not None: transaction[version]=c
+            if transaction is not None:
+                transaction[version] = c
             return c
 
         finally: self._r()
@@ -591,7 +591,8 @@
             d = {}
             for oid in storage.undo(id):
                 d[oid] = 1
-            self.invalidate(d)
+            # XXX I think we need to remove old undo to use mvcc
+            self.invalidate(None, d)
 
     def versionEmpty(self, version):
         return self._storage.versionEmpty(version)
@@ -619,13 +620,13 @@
 
     def commit(self, reallyme, t):
         dest=self._dest
-        oids = self._db._storage.commitVersion(self._version, dest, t)
+        tid, oids = self._db._storage.commitVersion(self._version, dest, t)
         oids = list2dict(oids)
-        self._db.invalidate(oids, version=dest)
+        self._db.invalidate(tid, oids, version=dest)
         if dest:
             # the code above just invalidated the dest version.
             # now we need to invalidate the source!
-            self._db.invalidate(oids, version=self._version)
+            self._db.invalidate(tid, oids, version=self._version)
 
 class AbortVersion(CommitVersion):
     """An object that will see to version abortion
@@ -634,9 +635,9 @@
     """
 
     def commit(self, reallyme, t):
-        version=self._version
-        oids = self._db._storage.abortVersion(version, t)
-        self._db.invalidate(list2dict(oids), version=version)
+        version = self._version
+        tid, oids = self._db._storage.abortVersion(version, t)
+        self._db.invalidate(tid, list2dict(oids), version=version)
 
 
 class TransactionalUndo(CommitVersion):
@@ -650,5 +651,5 @@
     # similarity of rhythm that I think it's justified.
 
     def commit(self, reallyme, t):
-        oids = self._db._storage.transactionalUndo(self._version, t)
-        self._db.invalidate(list2dict(oids))
+        tid, oids = self._db._storage.transactionalUndo(self._version, t)
+        self._db.invalidate(tid, list2dict(oids))


=== ZODB3/ZODB/Connection.py 1.100 => 1.100.2.1 ===
--- ZODB3/ZODB/Connection.py:1.100	Thu Oct  2 14:17:19 2003
+++ ZODB3/ZODB/Connection.py	Tue Oct  7 01:10:32 2003
@@ -21,7 +21,6 @@
 from POSException import ConflictError, ReadConflictError, TransactionError
 from ExtensionClass import Base
 import ExportImport, TmpStore
-from zLOG import LOG, ERROR, BLATHER, WARNING
 from coptimizations import new_persistent_id
 from ConflictResolution import ResolvedSerial
 from Transaction import Transaction, get_transaction
@@ -29,10 +28,17 @@
 
 from cPickle import Unpickler, Pickler
 from cStringIO import StringIO
+import logging
 import sys
 import threading
 from time import time
-from types import StringType, ClassType
+from types import ClassType
+
+_marker = object()
+
+def myhasattr(obj, attr):
+    # builtin hasattr() swallows exceptions
+    return getattr(obj, attr, _marker) is not _marker
 
 global_code_timestamp = 0
 
@@ -55,9 +61,9 @@
 
     The Connection manages movement of objects in and out of object storage.
     """
-    _tmp=None
-    _debug_info=()
-    _opened=None
+    _tmp = None
+    _debug_info = ()
+    _opened = None
     _code_timestamp = 0
     _transaction = None
 
@@ -65,9 +71,12 @@
     # when we close by putting something here.
 
     def __init__(self, version='', cache_size=400,
-                 cache_deactivate_after=60):
+                 cache_deactivate_after=60, mvcc=True):
         """Create a new Connection"""
-        self._version=version
+
+        self._log = logging.getLogger("zodb.conn")
+        
+        self._version = version
         self._cache = cache = PickleCache(self, cache_size)
         if version:
             # Caches for versions end up empty if the version
@@ -99,6 +108,14 @@
         self._invalidated = d = {}
         self._invalid = d.has_key
         self._conflicts = {}
+        self._noncurrent = {}
+
+        # If MVCC is enabled, then _mvcc is True and _txn_time stores
+        # the upper bound on transactions visible to this connection.
+        # That is, all object revisions must be written before _txn_time.
+        # If it is None, then the current revisions are acceptable.
+        self._mvcc = mvcc
+        self._txn_time = None
 
     def getTransaction(self):
         t = self._transaction
@@ -141,7 +158,7 @@
         self._incrgc = None
         self.cacheGC = None
 
-    def __getitem__(self, oid, tt=type(())):
+    def __getitem__(self, oid):
         obj = self._cache.get(oid, None)
         if obj is not None:
             return obj
@@ -157,9 +174,9 @@
 
         klass, args = object
 
-        if type(klass) is tt:
+        if isinstance(klass, tuple):
             module, name = klass
-            klass=self._db._classFactory(self, module, name)
+            klass = self._db._classFactory(self, module, name)
 
         if (args is None or
             not args and not hasattr(klass,'__getinitargs__')):
@@ -177,12 +194,10 @@
         self._cache[oid] = object
         return object
 
-    def _persistent_load(self,oid,
-                        tt=type(())):
-
+    def _persistent_load(self, oid):
         __traceback_info__=oid
 
-        if type(oid) is tt:
+        if isinstance(oid, tuple):
             # Quick instance reference.  We know all we need to know
             # to create the instance wo hitting the db, so go for it!
             oid, klass = oid
@@ -190,9 +205,10 @@
             if obj is not None:
                 return obj
 
-            if type(klass) is tt:
+            if isinstance(klass, tuple):
                 module, name = klass
-                try: klass=self._db._classFactory(self, module, name)
+                try:
+                    klass=self._db._classFactory(self, module, name)
                 except:
                     # Eek, we couldn't get the class. Hm.
                     # Maybe their's more current data in the
@@ -282,11 +298,12 @@
         # Call the close callbacks.
         if self.__onCloseCallbacks is not None:
             for f in self.__onCloseCallbacks:
-                try: f()
-                except:
-                    f=getattr(f, 'im_self', f)
-                    LOG('ZODB',ERROR, 'Close callback failed for %s' % f,
-                        error=sys.exc_info())
+                try:
+                    f()
+                except: # except what?
+                    f = getattr(f, 'im_self', f)
+                    self._log.error("Close callback failed for %s", f,
+                                    sys.exc_info())
             self.__onCloseCallbacks = None
         self._storage = self._tmp = self.new_oid = self._opened = None
         self._debug_info = ()
@@ -438,8 +455,8 @@
         if tmp is None: return
         src=self._storage
 
-        LOG('ZODB', BLATHER,
-            'Commiting subtransaction of size %s' % src.getSize())
+        self._log.debug("Commiting subtransaction of size %s",
+                        src.getSize())
 
         self._storage=tmp
         self._tmp=None
@@ -490,8 +507,6 @@
                 del o._p_jar
                 del o._p_oid
 
-    #XXX
-
     def db(self): return self._db
 
     def getVersion(self): return self._version
@@ -499,7 +514,7 @@
     def isReadOnly(self):
         return self._storage.isReadOnly()
 
-    def invalidate(self, oids):
+    def invalidate(self, tid, oids):
         """Invalidate a set of oids.
 
         This marks the oid as invalid, but doesn't actually invalidate
@@ -508,6 +523,8 @@
         """
         self._inv_lock.acquire()
         try:
+            if self._txn_time is None:
+                self._txn_time = tid
             self._invalidated.update(oids)
         finally:
             self._inv_lock.release()
@@ -517,13 +534,15 @@
         try:
             self._cache.invalidate(self._invalidated)
             self._invalidated.clear()
+            self._txn_time = None
         finally:
             self._inv_lock.release()
         # Now is a good time to collect some garbage
         self._cache.incrgc()
 
     def modifiedInVersion(self, oid):
-        try: return self._db.modifiedInVersion(oid)
+        try:
+            return self._db.modifiedInVersion(oid)
         except KeyError:
             return self._version
 
@@ -547,55 +566,85 @@
         if self._storage is None:
             msg = ("Shouldn't load state for %s "
                    "when the connection is closed" % oid_repr(oid))
-            LOG('ZODB', ERROR, msg)
+            self._log.error(msg)
             raise RuntimeError(msg)
 
         try:
-            # Avoid reading data from a transaction that committed
-            # after the current transaction started, as that might
-            # lead to mixing of cached data from earlier transactions
-            # and new inconsistent data.
-            #
-            # Wait for check until after data is loaded from storage
-            # to avoid time-of-check to time-of-use race.
-            p, serial = self._storage.load(oid, self._version)
-            self._load_count = self._load_count + 1
-            invalid = self._is_invalidated(obj)
-            self._set_ghost_state(obj, p)
-            obj._p_serial = serial
-            if invalid:
-                self._handle_independent(obj)
+            self._setstate(obj)
         except ConflictError:
             raise
         except:
-            LOG('ZODB', ERROR,
-                "Couldn't load state for %s" % oid_repr(oid),
-                error=sys.exc_info())
+            self._log.error("Couldn't load state for %s", oid_repr(oid),
+                            exc_info=sys.exc_info())
             raise
 
-    def _is_invalidated(self, obj):
-        # Helper method for setstate() covers three cases:
-        # returns false if obj is valid
-        # returns true if obj was invalidation, but is independent
-        # otherwise, raises ConflictError for invalidated objects
+    def _setstate(self, obj):
+        # Helper for setstate(), which provides logging of failures.
+
+        # The control flow is complicated here to avoid loading an
+        # object revision that we are sure we aren't going to use.  As
+        # a result, invalidation tests occur before and after the
+        # load.  We can only be sure about invalidations after the
+        # load.
+
+        # If an object has been invalidated, there are several cases
+        # to consider:
+        # 1. Check _p_independent()
+        # 2. Try MVCC
+        # 3. Raise ConflictError.
+
+        # Does anything actually use _p_independent()?  It would simplify
+        # the code if we could drop support for it.
+
+        # There is a harmless data race with self._invalidated.  A
+        # dict update could go on in another thread, but we don't care
+        # because we have to check again after the load anyway.
+        if (obj._p_oid in self._invalidated
+            and not myhasattr(obj, "_p_independent")):
+            # If the object has _p_independent(), we will handle it below.
+            if not (self._mvcc and self._setstate_noncurrent(obj)):
+                self.getTransaction().register(obj)
+                self._conflicts[obj._p_oid] = 1
+                raise ReadConflictError(object=obj)
+
+        p, serial = self._storage.load(obj._p_oid, self._version)
+        self._load_count += 1
+
         self._inv_lock.acquire()
         try:
-            if self._invalidated.has_key(obj._p_oid):
-                # Defer _p_independent() call until state is loaded.
-                ind = getattr(obj, "_p_independent", None)
-                if ind is not None:
-                    # Defer _p_independent() call until state is loaded.
-                    return 1
-                else:
-                    self.getTransaction().register(obj)
-                    self._conflicts[obj._p_oid] = 1
-                    raise ReadConflictError(object=obj)
-            else:
-                return 0
+            invalid = obj._p_oid in self._invalidated
         finally:
             self._inv_lock.release()
+            
+        if invalid:
+            if myhasattr(obj, "_p_independent"):
+                # This call will raise a ReadConflictError if something
+                # goes wrong
+                self._handle_independent(obj)
+            elif not (self._mvcc and self._setstate_noncurrent(obj)):
+                self.getTransaction().register(obj)
+                self._conflicts[obj._p_oid] = 1
+                raise ReadConflictError(object=obj)
+                
+        self._set_ghost_state(obj, p, serial)
+
+    def _setstate_noncurrent(self, obj):
+        """Set state using non-current data.
+
+        Return True if state was available, False if not.
+        """
+        try:
+            t = self._storage.loadNonCurrent(obj._p_oid, self._txn_time)
+        except KeyError:
+            return False
+        if t is None:
+            return False
+        data, serial, start, end = t
+        assert start < end == self._txn_time, (start, end, self._txn_time)
+        self._noncurrent[obj._p_oid] = True
+        self._set_ghost_state(obj, data, serial)
 
-    def _set_ghost_state(self, obj, p):
+    def _set_ghost_state(self, obj, p, serial):
         file = StringIO(p)
         unpickler = Unpickler(file)
         unpickler.persistent_load = self._persistent_load
@@ -607,6 +656,7 @@
             obj.update(state)
         else:
             setstate(state)
+        obj._p_serial = serial
 
     def _handle_independent(self, obj):
         # Helper method for setstate() handles possibly independent objects
@@ -649,9 +699,9 @@
             klass, args = copy
 
             if klass is not ExtensionKlass:
-                LOG('ZODB',ERROR,
-                    "Unexpected klass when setting class state on %s"
-                    % getattr(object,'__name__','(?)'))
+                self._log.error(
+                    "Unexpected klass when setting class state on %s",
+                    getattr(object, "__name__", "(?)"))
                 return
 
             copy = klass(*args)
@@ -663,7 +713,7 @@
             object._p_changed=0
             object._p_serial=serial
         except:
-            LOG('ZODB',ERROR, 'setklassstate failed', error=sys.exc_info())
+            self._log.error("setklassstate failed", exc_info=sys.exc_info())
             raise
 
     def tpc_abort(self, transaction):
@@ -720,7 +770,7 @@
 
         if not store_return:
             return
-        if isinstance(store_return, StringType):
+        if isinstance(store_return, str):
             assert oid is not None
             self._handle_one_serial(oid, store_return, change)
         else:
@@ -728,7 +778,7 @@
                 self._handle_one_serial(oid, serial, change)
 
     def _handle_one_serial(self, oid, serial, change):
-        if not isinstance(serial, StringType):
+        if not isinstance(serial, str):
             raise serial
         obj = self._cache.get(oid, None)
         if obj is None:
@@ -754,11 +804,11 @@
             self._storage._creating[:0]=self._creating
             del self._creating[:]
         else:
-            def callback():
+            def callback(tid):
                 d = {}
                 for oid in self._modified:
                     d[oid] = 1
-                self._db.invalidate(d, self)
+                self._db.invalidate(tid, d, self)
             self._storage.tpc_finish(transaction, callback)
 
         self._conflicts.clear()


=== ZODB3/ZODB/BaseStorage.py 1.36 => 1.36.2.1 ===
--- ZODB3/ZODB/BaseStorage.py:1.36	Thu Oct  2 14:17:19 2003
+++ ZODB3/ZODB/BaseStorage.py	Tue Oct  7 01:10:32 2003
@@ -56,12 +56,12 @@
     def abortVersion(self, src, transaction):
         if transaction is not self._transaction:
             raise POSException.StorageTransactionError(self, transaction)
-        return []
+        return self._serial, []
 
     def commitVersion(self, src, dest, transaction):
         if transaction is not self._transaction:
             raise POSException.StorageTransactionError(self, transaction)
-        return []
+        return self._serial, []
 
     def close(self):
         pass
@@ -199,7 +199,7 @@
                 return
             try:
                 if f is not None:
-                    f()
+                    f(self._serial)
                 u, d, e = self._ude
                 self._finish(self._serial, u, d, e)
                 self._clear_temp()
@@ -245,6 +245,33 @@
     def loadSerial(self, oid, serial):
         raise POSException.Unsupported, (
             "Retrieval of historical revisions is not supported")
+
+    def loadNonCurrent(self, oid, tid):
+        """Return most recent revision of oid before tid committed."""
+
+        n = 2
+        start_time = None
+        while start_time is None:
+            # The history() approach is a hack, because the dict
+            # returned by history() doesn't contain a tid.  It
+            # contains a serialno, which is often the same, but isn't
+            # required to be.  We'll pretend it is for now.
+
+            # A second problem is that history() doesn't say anything
+            # about whether the transaction status.  If it falls before
+            # the pack time, we can't honor the MVCC request.
+
+            # Note: history() returns the most recent record first.
+            L = self.history(oid, "", lambda d: not d["version"])
+            for d in L:
+                if d["serial"] < tid:
+                    start_time = d["serial"]
+                    break
+                else:
+                    end_time = d["serial"]
+            n *= 2
+        data = self.loadSerial(oid, start_time)
+        return data, start_time, start_time, end_time
 
     def getExtensionMethods(self):
         """getExtensionMethods




More information about the Zodb-checkins mailing list