DB.py.patch0100644000076500007650000000316707300005673012013 0ustar shaneshaneIndex: DB.py =================================================================== RCS file: /cvs-repository/Zope2/lib/python/ZODB/DB.py,v retrieving revision 1.28 diff -u -r1.28 DB.py --- DB.py 2001/04/19 16:06:25 1.28 +++ DB.py 2001/05/14 16:41:06 @@ -104,6 +104,7 @@ or more connections, which manage object spaces. Most of the actual work of managing objects is done by the connections. """ + klass = Connection def __init__(self, storage, pool_size=7, @@ -404,7 +405,7 @@ # This is a temporary connection. # We won't bother with the pools. This will be # a one-use connection. - c=Connection( + c=self.klass( version=version, cache_size=self._version_cache_size, cache_deactivate_after= @@ -455,7 +456,7 @@ c=None if version: if self._version_pool_size > len(allocated) or force: - c=Connection( + c=self.klass( version=version, cache_size=self._version_cache_size, cache_deactivate_after= @@ -463,7 +464,7 @@ allocated.append(c) pool.append(c) elif self._pool_size > len(allocated) or force: - c=Connection( + c=self.klass( version=version, cache_size=self._cache_size, cache_deactivate_after= ORConnection.py0100644000076500007650000003014007277412153012767 0ustar shaneshane from time import time from thread import allocate_lock from Connection import Connection from DB import DB from POSException import ConflictError ROOT_OID = '\0' * 8 NO_SERIAL = '\0' * 8 class ORMappingError (Exception): '''Invalid usage of object-relational mapping''' class ObjectMappingSpec: # Shared between threads. def getId(self): return id(self) def basicNew(self): # Called only on the root object for now. pass def loadState(self, object, key, jar): # Also calls jar.addGhostedChild() for each persistent subobject # or addDependentChild() for each object whose class is based on # Persistent but which should not be managed separately. # Implementations must modify the __dict__ rather than # set attributes directly! pass def storeState(self, object, key, jar, transaction): # Also calls jar.storeNewChild() for each new object. # It's best to read from the __dict__ only. pass class Jar: # One per persistent object. Never shared among threads. def __init__(self, connection, spec, key): self._c = connection # ORConnection self._spec = spec # ObjectMappingSpec instance or None for dependents self._key = key def getDBConnection(self): return self._c._dbconn def addGhostedChild(self, child, child_spec, child_key): ''' Invoked by ObjectMappingSpec. ''' # All ghosts that are children of a given object mapping # must have a different, hashable ID. It makes it possible to do # invalidation among multiple threads. # Loosely based on _persistent_load. oid = (child_spec.getId(), child_key) child._p_jar = Jar(self._c, child_spec, child_key) child._p_oid = oid child._p_changed = None # State not filled in. # object._p_serial = NO_SERIAL self._c._cache[oid] = child def addDependentChild(self, parent, child): ''' Invoked by ObjectMappingSpec. ''' # This persistent child will only notify the parent when it # is changed. child._p_jar = Jar(self._c, None, None) child._p_oid = None child._p_changed = 0 # State filled in. # We have to store the parent in the dict so that ZODB # can break the circular reference. # This would be handled better in cPersistence.c; we need # a special attribute called _p_parent which gets cleared # when the object's dict is cleared. child.__dict__['_p_parent'] = parent def storeNewChild(self, child, child_spec, child_key, transaction): ''' Invoked by ObjectMappingSpec. ''' # Loosely based on the commit method. oid = (child_spec.getId(), child_key) j = Jar(self._c, child_spec, child_key) child._p_jar = j child._p_oid = oid child._p_changed = 1 self._c._creating.append(oid) self._c._cache[oid] = child j.commit(child, transaction) def setstate(self, object): ''' Invoked by cPersistence. ''' if self._spec is None: raise ORMappingError, ('setstate() should not be called ' 'on a dependent child.') invalid = self._c._invalid oid = object._p_oid if invalid(oid) or invalid(None): if not hasattr(object.__class__, '_p_independent'): get_transaction().register(self) raise ConflictError(`oid`, `object.__class__`) invalid=1 else: invalid=0 self._spec.loadState(object, self._key, self) if invalid: if object._p_independent(): try: del self._c._invalidated[oid] except KeyError: pass else: get_transaction().register(self) raise ConflictError(`oid`, `object.__class__`) def commit(self, object, transaction): ''' Invoked by Transaction. ''' if self._spec is None: # A dependent child. Commit the parent instead. parent = object.__dict__['_p_parent'] hasattr(parent, 'id') # Activate the parent if necessary. transaction.register(parent) elif object._p_changed: invalid = self._c._invalid oid = object._p_oid if ( (invalid(oid) and not hasattr(object, '_p_resolveConflict')) or invalid(None) ): raise ConflictError, `oid` self._c._invalidating.append(oid) self._spec.storeState(object, self._key, self, transaction) def abort(self, object, transaction): self._c.abort(object, transaction) def root(self): return self._c.root() def tpc_abort(self, transaction): c = self._c if c._tpc_phase != 'abort': c._tpc_phase = 'abort' c.tpc_abort(transaction) def tpc_begin(self, transaction, sub=None): c = self._c if c._tpc_phase != 'begin': c._tpc_phase = 'begin' c.tpc_begin(transaction, sub) def tpc_vote(self, transaction): c = self._c if c._tpc_phase != 'vote': c._tpc_phase = 'vote' c.tpc_vote(transaction) def tpc_finish(self, transaction): c = self._c if c._tpc_phase != 'finish': c._tpc_phase = 'finish' c.tpc_finish(transaction) class ORConnection (Connection): # Generally one per thread. _tpc_phase = '' def __getitem__(self, oid): cache = self._cache if cache.has_key(oid): return cache[oid] if oid != ROOT_OID: raise 'Only the root object can be loaded this way' object = self._rootspec.basicNew() object._p_oid = oid object._p_jar = Jar(self, self._rootspec, oid) object._p_changed = None # State not filled in. # object._p_serial = NO_SERIAL cache[oid] = object return object def _setDB(self, odb): """Begin a new transaction. Any objects modified since the last transaction are invalidated. """ self._db = odb self._dbconn = odb._dbconn self._rootspec = odb._rootspec self._cache.invalidate(self._invalidated) self._opened=time() return self def close(self): self._incrgc() # This is a good time to do some GC db=self._db # Call the close callbacks. 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()) self.__onCloseCallbacks=() self._db=self._dbconn=self._rootspec=self._opened=None self._debug_info=() # Return the connection to the pool. db._closeConnection(self) def commit_sub(self, transaction): """Commit work done in subtransactions""" def abort_sub(self, transaction): """Abort work done in subtransactions""" self._cache.invalidate(src._index.keys()) self._invalidate_creating(src._creating) def tpc_abort(self, transaction): self.__onCommitActions = () dbconn = self._dbconn if hasattr(dbconn, 'tpc_abort'): dbconn.tpc_abort(transaction) cache=self._cache cache.invalidate(self._invalidated) cache.invalidate(self._invalidating) self._invalidate_creating() def tpc_begin(self, transaction, sub=None): if self._invalid(None): # Some nitwit invalidated everything! raise ConflictError, "transaction already invalidated" self._invalidating=[] self._creating=[] ## if sub: ## # Sub-transaction! ## _tmp=self._tmp ## if _tmp is None: ## _tmp=TmpStore.TmpStore(self._version) ## self._tmp=self._storage ## self._storage=_tmp ## _tmp.registerDB(self._db, 0) dbconn = self._dbconn if hasattr(dbconn, 'tpc_begin'): dbconn.tpc_begin(transaction) def tpc_vote(self, transaction, _type=type, _st=type('')): self.__onCommitActions = () dbconn = self._dbconn if hasattr(dbconn, 'tpc_vote'): s = dbconn.tpc_vote(transaction) if s: # Do we need this? get=self._cache.get for oid, s in s: o=get(oid, oid) if o is not oid: if _type(s) is not _st: raise s if s == ResolvedSerial: o._p_changed=None else: o._p_serial=s o._p_changed=0 def tpc_finish(self, transaction): # It's important that the storage call the function we pass # (self.tpc_finish_) while it still has it's lock. We don't # want another thread to be able to read any updated data # until we've had a chance to send an invalidation message to # all of the other connections! dbconn = self._dbconn if hasattr(dbconn, 'tpc_finish'): dbconn.tpc_finish(transaction, self._invalidate_invalidating) ## if self._tmp is not None: ## # Commiting a subtransaction! ## # There is no need to invalidate anything. ## self._storage.tpc_finish(transaction, self._invalidate_sub) ## self._storage._creating[:0]=self._creating ## del self._creating[:] ## else: ## self._storage.tpc_finish(transaction, ## self._invalidate_invalidating) self._cache.invalidate(self._invalidated) self._incrgc() # This is a good time to do some GC class ORDB (DB): klass = ORConnection def __init__(self, dbconn, rootspec, pool_size=7, cache_size=400, cache_deactivate_after=60, version_pool_size=3, version_cache_size=100, version_cache_deactivate_after=10, ): """Create an object database. The storage for the object database must be passed in. Optional arguments are: pool_size -- The size of the pool of object spaces. """ # Allocate locks: l=allocate_lock() self._a=l.acquire self._r=l.release # Setup connection pools and cache info self._pools={},[] self._temps=[] self._pool_size=pool_size self._cache_size=cache_size self._cache_deactivate_after=cache_deactivate_after self._version_pool_size=version_pool_size self._version_cache_size=version_cache_size self._version_cache_deactivate_after=version_cache_deactivate_after self._miv_cache={} # Setup storage self._dbconn = dbconn self._rootspec = rootspec if hasattr(dbconn, 'registerDB'): dbconn.registerDB(self, None) ## try: storage.load('\0\0\0\0\0\0\0\0','') ## except: ## import PersistentMapping ## file=cStringIO.StringIO() ## p=cPickle.Pickler(file,1) ## p.dump((PersistentMapping.PersistentMapping,None)) ## p.dump({'_container': {}}) ## t=Transaction() ## t.description='initial database creation' ## storage.tpc_begin(t) ## storage.store('\0\0\0\0\0\0\0\0', None, file.getvalue(), '', t) ## storage.tpc_vote(t) ## storage.tpc_finish(t) ## # Pass through methods: ## for m in ('history', ## 'supportsUndo', 'supportsVersions', 'undoLog', ## 'versionEmpty', 'versions'): ## setattr(self, m, getattr(storage, m)) ## if hasattr(storage, 'undoInfo'): ## self.undoInfo=storage.undoInfo def supportsUndo(self): return 0 def supportsVersions(self): return 0 testOR.py0100644000076500007650000000242307277412207011652 0ustar shaneshane import sys sys.path.append('..') import ZODB from ORConnection import ORDB, ObjectMappingSpec from PersistentMapping import PersistentMapping ROOT_OID = '\0' * 8 states = {ROOT_OID: PersistentMapping()} class PMSpec (ObjectMappingSpec): def basicNew(self): return PersistentMapping() def loadState(self, object, key, jar): s = states[key] d = {} for k, v in s.items(): d[k] = v if hasattr(v, '_p_oid'): jar.addGhostedChild(v, self, k) object.__dict__['_container'] = d def storeState(self, object, key, jar, transaction): # Also calls jar.storeNewChild() for each new object. # It's best to read from the __dict__ only. to_store = {} d = object.__dict__['_container'] for k, v in d.items(): if hasattr(v, '_p_oid'): if v._p_oid is None: print 'hi' jar.storeNewChild(v, self, k, transaction) to_store[k] = PersistentMapping() else: to_store[k] = v states[key] = to_store db = ORDB(None, PMSpec()) conn = db.open() root = conn.root() root['a'] = 1 root['b'] = PersistentMapping({5:6}) print states get_transaction().commit() print states