[Zodb-checkins] CVS: ZODB3/ZEO - ClientStorage.py:1.73.2.23.2.5
Jeremy Hylton
jeremy at zope.com
Wed Jun 11 15:42:54 EDT 2003
Update of /cvs-repository/ZODB3/ZEO
In directory cvs.zope.org:/tmp/cvs-serv824
Modified Files:
Tag: tim-loading_oids_status-branch
ClientStorage.py
Log Message:
Refactor handling of invalidation during load.
Keep count and set of invalidations in separate dicts keyed by oid.
=== ZODB3/ZEO/ClientStorage.py 1.73.2.23.2.4 => 1.73.2.23.2.5 ===
--- ZODB3/ZEO/ClientStorage.py:1.73.2.23.2.4 Wed Jun 11 12:34:39 2003
+++ ZODB3/ZEO/ClientStorage.py Wed Jun 11 14:42:54 2003
@@ -265,16 +265,16 @@
# 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 = {}
+ # first). To worm around this, load() creates a dict in
+ # _loading_oid_invs and invalidateTrans() stores any arriving
+ # invalidations in that dict. When the zeoLoad() call returns
+ # from the server, load() handles the invalidations.
+ # It's possible for different threads to attempt to load the same
+ # oid at the same time. To account for this, we keep a counter
+ # in _loading_oid_count. Invalidations are only handled when
+ # the count reaches zero.
+ self._loading_oid_count = {}
+ self._loading_oid_invs = {}
# Can't read data in one thread while writing data
# (tpc_finish) in another thread. In general, the lock
@@ -628,29 +628,28 @@
if self._server is None:
raise ClientDisconnected()
- self._incLoadStatus(oid, version)
+ self._incLoadStatus(oid)
try:
p, s, v, pv, sv = self._server.zeoLoad(oid)
except:
self._lock.acquire()
try:
- self._decLoadStatus(oid, version)
+ self._decLoadStatus(oid)
finally:
self._lock.release()
raise
self._lock.acquire()
try:
- statv, statnv = self._decLoadStatus(oid, version)
- if statv and statnv:
- self._cache.checkSize(0)
- self._cache.store(oid, p, s, v, pv, sv)
- else:
- if not statv:
- self._cache.invalidate(oid, version)
- if not statnv:
- self._cache.invalidate(oid, '')
+ is_last, invs = self._decLoadStatus(oid)
+ if is_last:
+ if invs:
+ for v in invs:
+ self._cache.invalidate(oid, v)
+ else:
+ self._cache.checkSize(0)
+ self._cache.store(oid, p, s, v, pv, sv)
finally:
self._lock.release()
@@ -661,45 +660,41 @@
return p, s
raise KeyError, oid # no non-version data for this
- def _incLoadStatus(self, oid, version):
+ def _incLoadStatus(self, oid):
"""Increment the load count for oid, version pair.
Does its own locking.
"""
- 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
+ count = self._loading_oid_count.get(oid)
+ if count is None:
+ count = 0
+ self._loading_oid_invs[oid] = {}
+ count += 1
+ self._loading_oid_count[oid] = count
finally:
self._lock.release()
- def _decLoadStatus(self, oid, version):
- """Decrement load count and return statii.
-
- Statii is a 2-element list of booleans indicating whether
- the object, version was invalidated during the load. The
- first element is oid, version; the second element is oid, "".
+ def _decLoadStatus(self, oid):
+ """Decrement load count.
+ Return boolean indicating whether this was the last load and
+ a list of versions to invalidate. The list is empty unless
+ the boolean is True.
+
Caller must hold self._lock.
"""
- statii = []
- pairs = (oid, version), (oid, "")
- 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]
- return statii
+ count = self._loading_oid_count[oid]
+ count -= 1
+ if count:
+ self._loading_oid_count[oid] = count
+ return 0, []
+ else:
+ del self._loading_oid_count[oid]
+ d = self._loading_oid_invs[oid]
+ del self._loading_oid_invs[oid]
+ return 1, d.keys()
def modifiedInVersion(self, oid):
"""Storage API: return the version, if any, that modfied an object.
@@ -997,17 +992,14 @@
d = versions.setdefault(version, {})
d[oid] = 1
- # Set invalidation flag for this (oid, version) pair.
- stuff = self._loading_oids_status.get(pair)
- if stuff:
- # load() is waiting for this. Keep the load count
- # the same, but clear the "no invalidations seen"
- # flag. This trick is necessary because zrpc invokes
- # invalidateTrans() immediately when an invalidation
- # arrives, but may not deliver the result of a load
- # until later, even if the load request preceded
- # the invalidation msg.
- self._loading_oids_status[pair] = stuff[0], 0
+ # Update the _loading_oids_invs dict for this oid,
+ # if necessary.
+ d = self._loading_oids_invs.get(oid)
+ if d is not None:
+ # load() is waiting for this. Mark the version
+ # as invalidated, so that load can invalidate it
+ # later.
+ d[version] = 1
else:
# load() isn't waiting for this. Simply invalidate it.
self._cache.invalidate(oid, version=version)
More information about the Zodb-checkins
mailing list