[Zodb-checkins] CVS: ZODB3/ZEO - ClientStorage.py:1.73.2.23.2.1

[Tim Peters]

Update of /cvs-repository/ZODB3/ZEO
In directory cvs.zope.org:/tmp/cvs-serv29657/ZEO

Modified Files:
      Tag: tim-loading_oids_status-branch
	ClientStorage.py 
Log Message:
Experimental code to worm around an out-of-order message problem in zrpc.


=== ZODB3/ZEO/ClientStorage.py 1.73.2.23 => 1.73.2.23.2.1 ===
--- ZODB3/ZEO/ClientStorage.py:1.73.2.23	Fri Jun  6 13:58:10 2003
+++ ZODB3/ZEO/ClientStorage.py	Tue Jun 10 18:57:12 2003
@@ -262,6 +262,20 @@
         self._oid_lock = threading.Lock()
         self._oids = [] # Object ids retrieved from new_oids()
 
+        # There's a nasty race.  The ZRPC layer can deliver invalidations
+        # out of order (i.e., the server sends the result of a load, then
+        # sends an invalidation for that object, but we see the invalidation
+        # first).  To worm around this, load() stores an (oid, version)
+        # pair for the requested object in the _loading_oids_status dict,
+        # mapping to a
+        #     (count, status)
+        # pair.  The count is the number of load() calls in progress
+        # that have requested this (oid, version) pair.  status is
+        # initially True, and a helper method of invalidateTrans() sets
+        # it False to record the invalidation.  load() then uses the
+        # status.  Note:  mutations are protected by self._lock.
+        self._loading_oids_status = {}
+
         # Can't read data in one thread while writing data
         # (tpc_finish) in another thread.  In general, the lock
         # must prevent access to the cache while _update_cache
@@ -614,13 +628,47 @@
         if self._server is None:
             raise ClientDisconnected()
 
-        # If an invalidation for oid comes in during zeoLoad, that's OK
-        # because we'll get oid's new state.
-
         # XXX Race condition among load / invalid / store in cache
-        p, s, v, pv, sv = self._server.zeoLoad(oid)
-        self._cache.checkSize(0)
-        self._cache.store(oid, p, s, v, pv, sv)
+        pairs = (oid, version), (oid, "")
+        self._lock.acquire()
+        try:
+            for pair in pairs:
+                stuff = self._loading_oids_status.get(pair)
+                if stuff is None:
+                    stuff = 1, 1    # count 1, status True
+                else:
+                    count, status = stuff
+                    stuff = count + 1, status
+                self._loading_oids_status[pair] = stuff
+        finally:
+            self._lock.release()
+
+        try:
+            p, s, v, pv, sv = self._server.zeoLoad(oid)
+        finally:
+            statii = [] # will be pair [version status, non-version status]
+            self._lock.acquire()
+            try:
+                for pair in pairs:
+                    count, status = self._loading_oids_status[pair]
+                    statii.append(status)
+                    count -= 1
+                    if count:
+                        self._loading_oids_status[pair] = count, status
+                    else:
+                        del self._loading_oids_status[pair]
+            finally:
+                self._lock.release()
+
+        if statii[0] and statii[1]: # both OK
+            self._cache.checkSize(0)
+            self._cache.store(oid, p, s, v, pv, sv)
+        else:
+            if not statii[0]:
+                self._cache.invalidate(oid, version)
+            if not statii[1]:
+                self._cache.invalidate(oid, '')
+
         if v and version and v == version:
             return pv, sv
         else:
@@ -919,10 +967,18 @@
         try:
             # versions maps version names to dictionary of invalidations
             versions = {}
-            for oid, version in invs:
+            for pair in invs:
+                oid, version = pair
                 d = versions.setdefault(version, {})
-                self._cache.invalidate(oid, version=version)
                 d[oid] = 1
+
+                # Set invalidation flag for this (oid, version) pair.
+                stuff = self._loading_oids_status.get(pair)
+                if stuff:
+                    self._loading_oids_status[pair] = stuff[0], 0
+                else:
+                    self._cache.invalidate(oid, version=version)
+
             if self._db is not None:
                 for v, d in versions.items():
                     self._db.invalidate(d, version=v)