[Zodb-checkins] CVS: ZODB3/ZEO - ClientStorage.py:1.110.2.2
Jeremy Hylton
cvs-admin at zope.org
Wed Nov 5 23:41:55 EST 2003
Update of /cvs-repository/ZODB3/ZEO
In directory cvs.zope.org:/tmp/cvs-serv29646/ZEO
Modified Files:
Tag: ZODB3-mvcc-2-branch
ClientStorage.py
Log Message:
Use the new, partial cache implementation.
This changes leaves several failing tests, because the cache isn't
persistent. It's just stored in a few dicts in memory. There are
four ZEO tests that depend on the cache persistent, and they all fail:
checkBasicPersistence
checkRollover
checkQuickVerificationWith2Clients
checkVerificationWith2ClientsInvqOverflow
The new cache implementation requires that the cache verification
protocols pass tids in addition to oids. The new code does that,
except that it passes None for tid on the cache verification. I have
to make sure the implications of that are fully worked out.
=== ZODB3/ZEO/ClientStorage.py 1.110.2.1 => 1.110.2.2 ===
--- ZODB3/ZEO/ClientStorage.py:1.110.2.1 Thu Oct 9 16:27:25 2003
+++ ZODB3/ZEO/ClientStorage.py Wed Nov 5 23:41:54 2003
@@ -26,7 +26,8 @@
import time
import types
-from ZEO import ClientCache, ServerStub
+from ZEO import ServerStub
+from ZEO.cache import Cache
from ZEO.TransactionBuffer import TransactionBuffer
from ZEO.Exceptions import ClientStorageError, UnrecognizedResult, \
ClientDisconnected, AuthError
@@ -91,7 +92,7 @@
# Classes we instantiate. A subclass might override.
TransactionBufferClass = TransactionBuffer
- ClientCacheClass = ClientCache.ClientCache
+ ClientCacheClass = Cache
ConnectionManagerClass = ConnectionManager
StorageServerStubClass = ServerStub.StorageServer
@@ -297,8 +298,7 @@
self._oid = '\0\0\0\0\0\0\0\0'
# Decide whether to use non-temporary files
- self._cache = self.ClientCacheClass(storage, cache_size,
- client=client, var=var)
+ self._cache = self.ClientCacheClass()
self._rpc_mgr = self.ConnectionManagerClass(addr, self,
tmin=min_disconnect_poll,
@@ -312,9 +312,6 @@
# doesn't succeed, call connect() to start a thread.
if not self._rpc_mgr.attempt_connect():
self._rpc_mgr.connect()
- # If the connect hasn't occurred, run with cached data.
- if not self._ready.isSet():
- self._cache.open()
def _wait(self, timeout=None):
if timeout is not None:
@@ -549,7 +546,6 @@
if ltid == last_inval_tid:
log2(INFO, "No verification necessary "
"(last_inval_tid up-to-date)")
- self._cache.open()
self._server = server
self._ready.set()
return "no verification"
@@ -563,7 +559,6 @@
pair = server.getInvalidations(last_inval_tid)
if pair is not None:
log2(INFO, "Recovering %d invalidations" % len(pair[1]))
- self._cache.open()
self.invalidateTransaction(*pair)
self._server = server
self._ready.set()
@@ -575,7 +570,9 @@
self._pickler = cPickle.Pickler(self._tfile, 1)
self._pickler.fast = 1 # Don't use the memo
- self._cache.verify(server.zeoVerify)
+ # XXX should batch these operations for efficiency
+ for oid, version, serial in self._cache.contents():
+ server.verify(oid, version, serial)
self._pending_server = server
server.endZeoVerify()
return "full verification"
@@ -739,25 +736,19 @@
finally:
self._lock.release()
- p, s, v, pv, sv = self._server.zeoLoad(oid)
+ data, serial, tid = self._server.loadEx(oid, version)
self._lock.acquire() # for atomic processing of invalidations
try:
if self._load_status:
- self._cache.checkSize(0)
- self._cache.store(oid, p, s, v, pv, sv)
+ self._cache.store(oid, version, serial, tid, None, data)
self._load_oid = None
finally:
self._lock.release()
finally:
self._load_lock.release()
- if v and version and v == version:
- return pv, sv
- else:
- if s:
- return p, s
- raise KeyError, oid # no non-version data for this
+ return data, serial
def modifiedInVersion(self, oid):
"""Storage API: return the version, if any, that modfied an object.
@@ -917,7 +908,7 @@
self._lock.acquire() # for atomic processing of invalidations
try:
- self._update_cache()
+ self._update_cache(tid)
if f is not None:
f(tid)
finally:
@@ -930,7 +921,7 @@
self._load_lock.release()
self.end_transaction()
- def _update_cache(self):
+ def _update_cache(self, tid):
"""Internal helper to handle objects modified by a transaction.
This iterates over the objects in the transaction buffer and
@@ -943,7 +934,6 @@
if self._cache is None:
return
- self._cache.checkSize(self._tbuf.get_size())
try:
self._tbuf.begin_iterate()
except ValueError, msg:
@@ -959,15 +949,13 @@
"client storage: %s" % msg)
if t is None:
break
- oid, v, p = t
- if p is None: # an invalidation
- s = None
- else:
+ oid, version, data = t
+ self._cache.invalidate(oid, version, tid)
+ # If data is None, we just invalidate.
+ if data is not None:
s = self._seriald[oid]
- if s == ResolvedSerial or s is None:
- self._cache.invalidate(oid, v)
- else:
- self._cache.update(oid, s, v, p)
+ if s != ResolvedSerial:
+ self._cache.store(oid, version, s, tid, None, data)
self._tbuf.clear()
def transactionalUndo(self, trans_id, trans):
@@ -985,19 +973,6 @@
self._tbuf.invalidate(oid, '')
return oids
- def undo(self, transaction_id):
- """Storage API: undo a transaction, writing directly to the storage."""
- if self._is_read_only:
- raise POSException.ReadOnlyError()
- oids = self._server.undo(transaction_id)
- self._lock.acquire()
- try:
- for oid in oids:
- self._cache.invalidate(oid, '')
- finally:
- self._lock.release()
- return oids
-
def undoInfo(self, first=0, last=-20, specification=None):
"""Storage API: return undo information."""
return self._server.undoInfo(first, last, specification)
@@ -1044,7 +1019,7 @@
return
self._pickler.dump(args)
- def _process_invalidations(self, tid, invs):
+ def _process_invalidations(self, invs):
# Invalidations are sent by the ZEO server as a sequence of
# oid, version pairs. The DB's invalidate() method expects a
# dictionary of oids.
@@ -1053,15 +1028,15 @@
try:
# versions maps version names to dictionary of invalidations
versions = {}
- for oid, version in invs:
+ for oid, version, tid in invs:
if oid == self._load_oid:
self._load_status = 0
- self._cache.invalidate(oid, version=version)
- versions.setdefault(version, {})[oid] = 1
+ self._cache.invalidate(oid, version, tid)
+ versions.setdefault((version, tid), {})[oid] = tid
if self._db is not None:
- for v, d in versions.items():
- self._db.invalidate(tid, d, version=v)
+ for (version, tid), d in versions.items():
+ self._db.invalidate(tid, d, version=version)
finally:
self._lock.release()
@@ -1093,7 +1068,8 @@
for t in args:
self._pickler.dump(t)
return
- self._process_invalidations(tid, args)
+ self._process_invalidations([(oid, version, tid)
+ for oid, version in args])
# The following are for compatibility with protocol version 2.0.0
@@ -1104,36 +1080,10 @@
end = endVerify
Invalidate = invalidateTrans
-try:
- StopIteration
-except NameError:
- class StopIteration(Exception):
- pass
-
-class InvalidationLogIterator:
- """Helper class for reading invalidations in endVerify."""
-
- def __init__(self, fileobj):
- self._unpickler = cPickle.Unpickler(fileobj)
- self.getitem_i = 0
-
- def __iter__(self):
- return self
-
- def next(self):
- oid, version = self._unpickler.load()
+def InvalidationLogIterator(fileobj):
+ unpickler = cPickle.Unpickler(fileobj)
+ while 1:
+ oid, version = unpickler.load()
if oid is None:
- raise StopIteration
- return oid, version
-
- # The __getitem__() method is needed to support iteration
- # in Python 2.1.
-
- def __getitem__(self, i):
- assert i == self.getitem_i
- try:
- obj = self.next()
- except StopIteration:
- raise IndexError, i
- self.getitem_i += 1
- return obj
+ break
+ yield oid, version, None
More information about the Zodb-checkins
mailing list