SVN: ZODB/branches/jim-thready/src/Z rebranch
Jim Fulton <[email protected]>
| Newsgroups | gmane.comp.python.zope.zodb.cvs |
|---|---|
| Message-ID | <[email protected]> |
Log message for revision 108162:
rebranch
Changed:
U ZODB/branches/jim-thready/src/ZEO/ClientStorage.py
U ZODB/branches/jim-thready/src/ZEO/CommitLog.py
U ZODB/branches/jim-thready/src/ZEO/ServerStub.py
U ZODB/branches/jim-thready/src/ZEO/StorageServer.py
U ZODB/branches/jim-thready/src/ZEO/tests/CommitLockTests.py
U ZODB/branches/jim-thready/src/ZEO/tests/InvalidationTests.py
U ZODB/branches/jim-thready/src/ZEO/tests/servertesting.py
U ZODB/branches/jim-thready/src/ZEO/tests/testConversionSupport.py
U ZODB/branches/jim-thready/src/ZEO/tests/testZEO.py
U ZODB/branches/jim-thready/src/ZEO/tests/testZEO2.py
U ZODB/branches/jim-thready/src/ZEO/zrpc/client.py
U ZODB/branches/jim-thready/src/ZEO/zrpc/connection.py
U ZODB/branches/jim-thready/src/ZEO/zrpc/trigger.py
U ZODB/branches/jim-thready/src/ZODB/Connection.py
U ZODB/branches/jim-thready/src/ZODB/FileStorage/FileStorage.py
U ZODB/branches/jim-thready/src/ZODB/FileStorage/format.py
U ZODB/branches/jim-thready/src/ZODB/tests/RevisionStorage.py
U ZODB/branches/jim-thready/src/ZODB/tests/StorageTestBase.py
U ZODB/branches/jim-thready/src/ZODB/tests/testFileStorage.py
-=-
Modified: ZODB/branches/jim-thready/src/ZEO/ClientStorage.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/ClientStorage.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/ClientStorage.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -1198,11 +1198,13 @@
if self._cache is None:
return
- for oid, data in self._tbuf:
+ for oid, tid in self._seriald.iteritems():
self._cache.invalidate(oid, tid, False)
+
+ for oid, data in self._tbuf:
+ s = self._seriald[oid] # assigning here asserts that oid in seriald
# If data is None, we just invalidate.
if data is not None:
- s = self._seriald[oid]
if s != ResolvedSerial:
assert s == tid, (s, tid)
self._cache.store(oid, s, None, data)
@@ -1241,10 +1243,7 @@
"""
self._check_trans(txn)
- tid, oids = self._server.undo(trans_id, id(txn))
- for oid in oids:
- self._tbuf.invalidate(oid)
- return tid, oids
+ self._server.undoa(trans_id, id(txn))
def undoInfo(self, first=0, last=-20, specification=None):
"""Storage API: return undo information."""
Modified: ZODB/branches/jim-thready/src/ZEO/CommitLog.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/CommitLog.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/CommitLog.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -32,27 +32,31 @@
self.pickler = cPickle.Pickler(self.file, 1)
self.pickler.fast = 1
self.stores = 0
- self.read = 0
def size(self):
return self.file.tell()
def delete(self, oid, serial):
- self.pickler.dump(('d', oid, serial))
+ self.pickler.dump(('_delete', (oid, serial)))
self.stores += 1
def store(self, oid, serial, data):
- self.pickler.dump(('s', oid, serial, data))
+ self.pickler.dump(('_store', (oid, serial, data)))
self.stores += 1
def restore(self, oid, serial, data, prev_txn):
- self.pickler.dump(('r', oid, serial, data, prev_txn))
+ self.pickler.dump(('_restore', (oid, serial, data, prev_txn)))
self.stores += 1
- def get_loader(self):
- self.read = 1
+ def undo(self, transaction_id):
+ self.pickler.dump(('_undo', (transaction_id, )))
+ self.stores += 1
+
+ def __iter__(self):
self.file.seek(0)
- return self.stores, cPickle.Unpickler(self.file)
+ unpickler = cPickle.Unpickler(self.file)
+ for i in range(self.stores):
+ yield unpickler.load()
def close(self):
if self.file:
Modified: ZODB/branches/jim-thready/src/ZEO/ServerStub.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/ServerStub.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/ServerStub.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -272,8 +272,8 @@
def new_oid(self):
return self.rpc.call('new_oid')
- def undo(self, trans_id, trans):
- return self.rpc.call('undo', trans_id, trans)
+ def undoa(self, trans_id, trans):
+ self.rpc.callAsync('undoa', trans_id, trans)
def undoLog(self, first, last):
return self.rpc.call('undoLog', first, last)
Modified: ZODB/branches/jim-thready/src/ZEO/StorageServer.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/StorageServer.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/StorageServer.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -20,6 +20,8 @@
exported for invocation by the server.
"""
+from __future__ import with_statement
+
import asyncore
import cPickle
import logging
@@ -32,6 +34,7 @@
import transaction
+import ZODB.blob
import ZODB.serialize
import ZODB.TimeStamp
import ZEO.zrpc.error
@@ -48,7 +51,7 @@
from ZODB.POSException import StorageError, StorageTransactionError
from ZODB.POSException import TransactionError, ReadOnlyError, ConflictError
from ZODB.serialize import referencesf
-from ZODB.utils import u64, p64, oid_repr, mktemp
+from ZODB.utils import u64, p64, oid_repr
from ZODB.loglevels import BLATHER
@@ -87,7 +90,6 @@
def __init__(self, server, read_only=0, auth_realm=None):
self.server = server
# timeout and stats will be initialized in register()
- self.timeout = None
self.stats = None
self.connection = None
self.client = None
@@ -95,14 +97,13 @@
self.storage_id = "uninitialized"
self.transaction = None
self.read_only = read_only
- self.locked = 0
+ self.locked = False # Don't have storage lock
self.verifying = 0
self.store_failed = 0
self.log_label = _label
self.authenticated = 0
self.auth_realm = auth_realm
self.blob_tempfile = None
- self.blob_log = []
# The authentication protocol may define extra methods.
self._extensions = {}
for func in self.extensions:
@@ -139,24 +140,13 @@
self.log_label = _label + "/" + label
def notifyDisconnected(self):
+ self.connection = None
+
# When this storage closes, we must ensure that it aborts
# any pending transaction.
if self.transaction is not None:
self.log("disconnected during transaction %s" % self.transaction)
- if not self.locked:
- # Delete (d, zeo_storage) from the _waiting list, if found.
- waiting = self.storage._waiting
- for i in range(len(waiting)):
- d, z = waiting[i]
- if z is self:
- del waiting[i]
- self.log("Closed connection removed from waiting list."
- " Clients waiting: %d." % len(waiting))
- break
-
- if self.transaction:
- self.tpc_abort(self.transaction.id)
-
+ self.tpc_abort(self.transaction.id)
else:
self.log("disconnected")
@@ -176,6 +166,7 @@
def setup_delegation(self):
"""Delegate several methods to the storage
"""
+ # Called from register
storage = self.storage
@@ -183,9 +174,6 @@
if not info['supportsUndo']:
self.undoLog = self.undoInfo = lambda *a,**k: ()
- def undo(*a, **k):
- raise NotImplementedError
- self.undo = undo
self.getTid = storage.getTid
self.load = storage.load
@@ -265,9 +253,12 @@
if self.auth_realm and not self.authenticated:
raise AuthError("Client was never authenticated with server!")
+ self.connection.auth_done()
+
if self.storage is not None:
self.log("duplicate register() call")
raise ValueError("duplicate register() call")
+
storage = self.server.storages.get(storage_id)
if storage is None:
self.log("unknown storage_id: %s" % storage_id)
@@ -280,19 +271,15 @@
self.storage_id = storage_id
self.storage = storage
self.setup_delegation()
- self.timeout, self.stats = self.server.register_connection(storage_id,
- self)
+ self.stats = self.server.register_connection(storage_id, self)
def get_info(self):
storage = self.storage
- try:
- supportsUndo = storage.supportsUndo
- except AttributeError:
- supportsUndo = False
- else:
- supportsUndo = supportsUndo()
+ supportsUndo = (getattr(storage, 'supportsUndo', lambda : False)()
+ and self.connection.peer_protocol_version >= 'Z310')
+
# Communicate the backend storage interfaces to the client
storage_provides = zope.interface.providedBy(storage)
interfaces = []
@@ -419,6 +406,7 @@
self.serials = []
self.invalidated = []
self.txnlog = CommitLog()
+ self.blob_log = []
self.tid = tid
self.status = status
self.store_failed = 0
@@ -438,131 +426,101 @@
if not self._check_tid(id):
return
assert self.locked
+
self.stats.commits += 1
- self.storage.tpc_finish(self.transaction)
+ self.storage.tpc_finish(self.transaction, self._invalidate)
+ # Note that the tid is still current because we still hold the
+ # commit lock. We'll relinquish it in _clear_transaction.
tid = self.storage.lastTransaction()
- if self.invalidated:
- self.server.invalidate(self, self.storage_id, tid,
- self.invalidated, self.get_size_info())
self._clear_transaction()
# Return the tid, for cache invalidation optimization
return tid
- def tpc_abort(self, id):
- if not self._check_tid(id):
+ def _invalidate(self, tid):
+ if self.invalidated:
+ self.server.invalidate(self, self.storage_id, tid,
+ self.invalidated, self.get_size_info())
+
+ def tpc_abort(self, tid):
+ if not self._check_tid(tid):
return
self.stats.aborts += 1
+
+ # Is there a race here? What if notifyLocked is called after
+ # the check? Well, we still won't have started committing the actual
+ # storage. That wouldn't happen until _restart is called and that
+ # can't happen while this method is executing, as both are only
+ # run by the client thtread. So no race.
if self.locked:
self.storage.tpc_abort(self.transaction)
self._clear_transaction()
def _clear_transaction(self):
# Common code at end of tpc_finish() and tpc_abort()
- self.stats.active_txns -= 1
- self.transaction = None
- self.txnlog.close()
if self.locked:
+ self.server.unlock_storage(self)
self.locked = 0
- self.timeout.end(self)
- self.stats.lock_time = None
- self.log("Transaction released storage lock", BLATHER)
+ self.transaction = None
+ self.stats.active_txns -= 1
+ if self.txnlog is not None:
+ self.txnlog.close()
+ self.txnlog = None
+ for oid, oldserial, data, blobfilename in self.blob_log:
+ ZODB.blob.remove_committed(blobfilename)
+ del self.blob_log
- # Restart any client waiting for the storage lock.
- while self.storage._waiting:
- delay, zeo_storage = self.storage._waiting.pop(0)
- try:
- zeo_storage._restart(delay)
- except:
- self.log("Unexpected error handling waiting transaction",
- level=logging.WARNING, exc_info=True)
- zeo_storage.connection.close()
- continue
+ def vote(self, tid):
+ self._check_tid(tid, exc=StorageTransactionError)
+ return self._try_to_vote()
- if self.storage._waiting:
- n = len(self.storage._waiting)
- self.log("Blocked transaction restarted. "
- "Clients waiting: %d" % n)
+ def _try_to_vote(self, delay=None):
+ if self.connection is None:
+ return # We're disconnected
+ self.locked = self.server.lock_storage(self)
+ if self.locked:
+ self.log("(%r) unlock: transactions waiting: %s"
+ % (self.storage_id, self.server.waiting(self)))
+ try:
+ self._vote()
+ except Exception:
+ if delay is not None:
+ delay.error()
else:
- self.log("Blocked transaction restarted.")
-
- break
-
- # The following two methods return values, so they must acquire
- # the storage lock and begin the transaction before returning.
-
- # It's a bit vile that undo can cause us to get the lock before vote.
-
- def undo(self, trans_id, id):
- self._check_tid(id, exc=StorageTransactionError)
- if self.locked:
- return self._undo(trans_id)
+ raise
+ else:
+ if delay is not None:
+ delay.reply(None)
else:
- return self._wait(lambda: self._undo(trans_id))
+ if delay == None:
+ self.log("(%r) queue lock: transactions waiting: %s"
+ % (self.storage_id, self.server.waiting(self)+1))
+ delay = Delay()
+ self.server.unlock_callback(self, delay)
+ return delay
- def vote(self, id):
- self._check_tid(id, exc=StorageTransactionError)
- if self.locked:
- return self._vote()
- else:
- return self._wait(lambda: self._vote())
+ def _unlock_callback(self, delay):
+ connection = self.connection
+ if connection is not None:
+ connection.trigger.pull_trigger(self._try_to_vote, delay)
- # When a delayed transaction is restarted, the dance is
- # complicated. The restart occurs when one ZEOStorage instance
- # finishes as a transaction and finds another instance is in the
- # _waiting list.
+ def _vote(self):
- # It might be better to have a mechanism to explicitly send
- # the finishing transaction's reply before restarting the waiting
- # transaction. If the restart takes a long time, the previous
- # client will be blocked until it finishes.
-
- def _wait(self, thunk):
- # Wait for the storage lock to be acquired.
- self._thunk = thunk
- if self.tpc_transaction():
- d = Delay()
- self.storage._waiting.append((d, self))
- self.log("Transaction blocked waiting for storage. "
- "Clients waiting: %d." % len(self.storage._waiting))
- return d
- else:
- self.log("Transaction acquired storage lock.", BLATHER)
- return self._restart()
-
- def _restart(self, delay=None):
- # Restart when the storage lock is available.
if self.txnlog.stores == 1:
template = "Preparing to commit transaction: %d object, %d bytes"
else:
template = "Preparing to commit transaction: %d objects, %d bytes"
+
self.log(template % (self.txnlog.stores, self.txnlog.size()),
level=BLATHER)
- self.locked = 1
- self.timeout.begin(self)
- self.stats.lock_time = time.time()
if (self.tid is not None) or (self.status != ' '):
self.storage.tpc_begin(self.transaction, self.tid, self.status)
else:
self.storage.tpc_begin(self.transaction)
try:
- loads, loader = self.txnlog.get_loader()
- for i in range(loads):
- store = loader.load()
- store_type = store[0]
- store_args = store[1:]
-
- if store_type == 'd':
- do_store = self._delete
- elif store_type == 's':
- do_store = self._store
- elif store_type == 'r':
- do_store = self._restore
- else:
- raise ValueError('Invalid store type: %r' % store_type)
-
- if not do_store(*store_args):
+ for op, args in self.txnlog:
+ if not getattr(self, op)(*args):
break
# Blob support
@@ -575,12 +533,17 @@
self._clear_transaction()
raise
- resp = self._thunk()
- if delay is not None:
- delay.reply(resp)
- else:
- return resp
+ if not self.store_failed:
+ # Only call tpc_vote of no store call failed, otherwise
+ # the serialnos() call will deliver an exception that will be
+ # handled by the client in its tpc_vote() method.
+ serials = self.storage.tpc_vote(self.transaction)
+ if serials:
+ self.serials.extend(serials)
+
+ self.client.serialnos(self.serials)
+
# The public methods of the ZEO client API do not do the real work.
# They defer work until after the storage lock has been acquired.
# Most of the real implementations are in methods beginning with
@@ -610,14 +573,18 @@
os.write(self.blob_tempfile[0], chunk)
def storeBlobEnd(self, oid, serial, data, id):
+ self._check_tid(id, exc=StorageTransactionError)
+ assert self.txnlog is not None # effectively not allowed after undo
fd, tempname = self.blob_tempfile
self.blob_tempfile = None
os.close(fd)
self.blob_log.append((oid, serial, data, tempname))
def storeBlobShared(self, oid, serial, data, filename, id):
+ self._check_tid(id, exc=StorageTransactionError)
+ assert self.txnlog is not None # effectively not allowed after undo
+
# Reconstruct the full path from the filename in the OID directory
-
if (os.path.sep in filename
or not (filename.endswith('.tmp')
or filename[:-1].endswith('.tmp')
@@ -635,6 +602,13 @@
def sendBlob(self, oid, serial):
self.client.storeBlob(oid, serial, self.storage.loadBlob(oid, serial))
+ def undo(*a, **k):
+ raise NotImplementedError
+
+ def undoa(self, trans_id, tid):
+ self._check_tid(tid, exc=StorageTransactionError)
+ self.txnlog.undo(trans_id)
+
def _delete(self, oid, serial):
err = None
try:
@@ -721,6 +695,27 @@
return err is None
+ def _undo(self, trans_id):
+ err = None
+ try:
+ tid, oids = self.storage.undo(trans_id, self.transaction)
+ except (SystemExit, KeyboardInterrupt):
+ raise
+ except Exception, err:
+ self.store_failed = 1
+ if not isinstance(err, TransactionError):
+ # Unexpected errors are logged and passed to the client
+ self.log("store error: %s, %s" % sys.exc_info()[:2],
+ logging.ERROR, exc_info=True)
+ err = self._marshal_error(err)
+ # The exception is reported back as newserial for this oid
+ self.serials.append((oid, err))
+ else:
+ self.invalidated.extend(oids)
+ self.serials.extend((oid, ResolvedSerial) for oid in oids)
+
+ return err is None
+
def _marshal_error(self, error):
# Try to pickle the exception. If it can't be pickled,
# the RPC response would fail, so use something that can be pickled.
@@ -734,23 +729,6 @@
error = StorageServerError(msg)
return error
- def _vote(self):
- if not self.store_failed:
- # Only call tpc_vote of no store call failed, otherwise
- # the serialnos() call will deliver an exception that will be
- # handled by the client in its tpc_vote() method.
- serials = self.storage.tpc_vote(self.transaction)
- if serials:
- self.serials.extend(serials)
-
- self.client.serialnos(self.serials)
- return
-
- def _undo(self, trans_id):
- tid, oids = self.storage.undo(trans_id, self.transaction)
- self.invalidated.extend(oids)
- return tid, oids
-
# IStorageIteration support
def iterator_start(self, start, stop):
@@ -929,8 +907,12 @@
for name, storage in storages.items()])
log("%s created %s with storages: %s" %
(self.__class__.__name__, read_only and "RO" or "RW", msg))
- for s in storages.values():
- s._waiting = []
+
+
+ self._lock = threading.Lock()
+ self._commit_locks = {}
+ self._unlock_callbacks = dict((name, []) for name in storages)
+
self.read_only = read_only
self.auth_protocol = auth_protocol
self.auth_database = auth_database
@@ -1044,7 +1026,7 @@
Returns the timeout and stats objects for the appropriate storage.
"""
self.connections[storage_id].append(conn)
- return self.timeouts[storage_id], self.stats[storage_id]
+ return self.stats[storage_id]
def _invalidateCache(self, storage_id):
"""We need to invalidate any caches we have.
@@ -1216,7 +1198,46 @@
if conn.obj in cl:
cl.remove(conn.obj)
+ def lock_storage(self, zeostore):
+ storage_id = zeostore.storage_id
+ with self._lock:
+ if storage_id in self._commit_locks:
+ return False
+ self._commit_locks[storage_id] = zeostore
+ self.timeouts[storage_id].begin(zeostore)
+ self.stats[storage_id].lock_time = time.time()
+ return True
+ def unlock_storage(self, zeostore):
+ storage_id = zeostore.storage_id
+ with self._lock:
+ assert self._commit_locks[storage_id] is zeostore
+ del self._commit_locks[storage_id]
+ self.timeouts[storage_id].end(zeostore)
+ self.stats[storage_id].lock_time = None
+ callbacks = self._unlock_callbacks[storage_id][:]
+ del self._unlock_callbacks[storage_id][:]
+
+ if callbacks:
+ zeostore.log("(%r) unlock: transactions waiting: %s"
+ % (storage_id, len(callbacks)-1))
+
+ for zeostore, delay in callbacks:
+ try:
+ zeostore._unlock_callback(delay)
+ except (SystemExit, KeyboardInterrupt):
+ raise
+ except Exception:
+ logger.exception("Calling unlock callback")
+
+ def unlock_callback(self, zeostore, delay):
+ storage_id = zeostore.storage_id
+ with self._lock:
+ self._unlock_callbacks[storage_id].append((zeostore, delay))
+
+ def waiting(self, zeostore):
+ return len(self._unlock_callbacks[zeostore.storage_id])
+
class StubTimeoutThread:
def begin(self, client):
@@ -1238,7 +1259,6 @@
self._client = None
self._deadline = None
self._cond = threading.Condition() # Protects _client and _deadline
- self._trigger = trigger()
def begin(self, client):
# Called from the restart code the "main" thread, whenever the
@@ -1281,7 +1301,7 @@
if howlong <= 0:
client.log("Transaction timeout after %s seconds" %
self._timeout)
- self._trigger.pull_trigger(lambda: client.connection.close())
+ client.connection.trigger.pull_trigger(client.connection.close)
else:
time.sleep(howlong)
@@ -1337,7 +1357,7 @@
self.rpc.callAsync('endVerify')
def invalidateTransaction(self, tid, args):
- self.rpc.callAsyncNoPoll('invalidateTransaction', tid, args)
+ self.rpc.callAsync('invalidateTransaction', tid, args)
def serialnos(self, arg):
self.rpc.callAsync('serialnos', arg)
@@ -1363,7 +1383,7 @@
class ClientStub308(ClientStub):
def invalidateTransaction(self, tid, args):
- self.rpc.callAsyncNoPoll(
+ self.rpc.callAsync(
'invalidateTransaction', tid, [(arg, '') for arg in args])
def invalidateVerify(self, oid):
Modified: ZODB/branches/jim-thready/src/ZEO/tests/CommitLockTests.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/tests/CommitLockTests.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/tests/CommitLockTests.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -181,64 +181,3 @@
self._finish_threads()
self._cleanup()
-
-class CommitLockUndoTests(CommitLockTests):
-
- def _get_trans_id(self):
- self._dostore()
- L = self._storage.undoInfo()
- return L[0]['id']
-
- def _begin_undo(self, trans_id, txn):
- rpc = self._storage._server.rpc
- return rpc._deferred_call('undo', trans_id, id(txn))
-
- def _finish_undo(self, msgid):
- return self._storage._server.rpc._deferred_wait(msgid)
-
- def checkCommitLockUndoFinish(self):
- trans_id = self._get_trans_id()
- oid, txn = self._start_txn()
- msgid = self._begin_undo(trans_id, txn)
-
- self._begin_threads()
-
- self._finish_undo(msgid)
- self._storage.tpc_vote(txn)
- self._storage.tpc_finish(txn)
- self._storage.load(oid, '')
-
- self._finish_threads()
-
- self._dostore()
- self._cleanup()
-
- def checkCommitLockUndoAbort(self):
- trans_id = self._get_trans_id()
- oid, txn = self._start_txn()
- msgid = self._begin_undo(trans_id, txn)
-
- self._begin_threads()
-
- self._finish_undo(msgid)
- self._storage.tpc_vote(txn)
- self._storage.tpc_abort(txn)
-
- self._finish_threads()
-
- self._dostore()
- self._cleanup()
-
- def checkCommitLockUndoClose(self):
- trans_id = self._get_trans_id()
- oid, txn = self._start_txn()
- msgid = self._begin_undo(trans_id, txn)
- self._begin_threads()
-
- self._finish_undo(msgid)
- self._storage.tpc_vote(txn)
- self._storage.close()
-
- self._finish_threads()
-
- self._cleanup()
Modified: ZODB/branches/jim-thready/src/ZEO/tests/InvalidationTests.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/tests/InvalidationTests.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/tests/InvalidationTests.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -318,9 +318,9 @@
# tearDown then immediately, but if other threads are still
# running that can lead to a cascade of spurious exceptions.
for t in threads:
- t.join(10)
+ t.join(30)
for t in threads:
- t.cleanup()
+ t.cleanup(10)
def checkConcurrentUpdates2Storages_emulated(self):
self._storage = storage1 = self.openClientStorage()
@@ -378,6 +378,34 @@
db1.close()
db2.close()
+ def checkConcurrentUpdates19Storages(self):
+ n = 19
+ dbs = [DB(self.openClientStorage()) for i in range(n)]
+ self._storage = dbs[0].storage
+ stop = threading.Event()
+
+ cn = dbs[0].open()
+ tree = cn.root()["tree"] = OOBTree()
+ transaction.commit()
+ cn.close()
+
+ # Run threads that update the BTree
+ cd = {}
+ threads = [self.StressThread(dbs[i], stop, i, cd, i, n)
+ for i in range(n)]
+ self.go(stop, cd, *threads)
+
+ while len(set(db.lastTransaction() for db in dbs)) > 1:
+ _ = [db._storage.sync() for db in dbs]
+
+ cn = dbs[0].open()
+ tree = cn.root()["tree"]
+ self._check_tree(cn, tree)
+ self._check_threads(tree, *threads)
+
+ cn.close()
+ _ = [db.close() for db in dbs]
+
def checkConcurrentUpdates1Storage(self):
self._storage = storage1 = self.openClientStorage()
db1 = DB(storage1)
Modified: ZODB/branches/jim-thready/src/ZEO/tests/servertesting.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/tests/servertesting.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/tests/servertesting.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -56,3 +56,13 @@
def callAsync(self, meth, *args):
print self.name, 'callAsync', meth, repr(args)
+
+ @property
+ def trigger(self):
+ return self
+
+ def pull_trigger(self, func):
+ func()
+
+ def auth_done(self):
+ pass
Modified: ZODB/branches/jim-thready/src/ZEO/tests/testConversionSupport.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/tests/testConversionSupport.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/tests/testConversionSupport.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -13,6 +13,8 @@
##############################################################################
import unittest
from zope.testing import doctest
+import ZEO.zrpc.connection
+import ZEO.tests.servertesting
class FakeStorageBase:
@@ -29,6 +31,9 @@
def __len__(self):
return 4
+ def registerDB(self, *args):
+ pass
+
class FakeStorage(FakeStorageBase):
def record_iternext(self, next=None):
@@ -50,15 +55,25 @@
def register_connection(*args):
return None, None
+class FauxConn:
+ addr = 'x'
+ thread_ident = unregistered_thread_ident = None
+ peer_protocol_version = (
+ ZEO.zrpc.connection.Connection.current_protocol)
+
+ def auth_done(self):
+ pass
+
def test_server_record_iternext():
"""
-
+
On the server, record_iternext calls are simply delegated to the
underlying storage.
>>> import ZEO.StorageServer
>>> zeo = ZEO.StorageServer.ZEOStorage(FakeServer(), False)
+ >>> zeo.notifyConnected(FauxConn())
>>> zeo.register('1', False)
>>> next = None
@@ -71,13 +86,14 @@
2
3
4
-
+
The storage info also reflects the fact that record_iternext is supported.
>>> zeo.get_info()['supports_record_iternext']
True
>>> zeo = ZEO.StorageServer.ZEOStorage(FakeServer(), False)
+ >>> zeo.notifyConnected(FauxConn())
>>> zeo.register('2', False)
>>> zeo.get_info()['supports_record_iternext']
@@ -152,7 +168,7 @@
4
"""
-
+
def history_to_version_compatible_storage():
"""
Some storages work under ZODB <= 3.8 and ZODB >= 3.9.
@@ -163,15 +179,19 @@
... return oid,version,size
A ZEOStorage such as the following should support this type of storage:
-
+
>>> class OurFakeServer(FakeServer):
... storages = {'1':VersionCompatibleStorage()}
>>> import ZEO.StorageServer
- >>> zeo = ZEO.StorageServer.ZEOStorage(OurFakeServer(), False)
+ >>> zeo = ZEO.StorageServer.ZEOStorage(
+ ... ZEO.tests.servertesting.StorageServer(
+ ... 'test', {'1':VersionCompatibleStorage()}))
+ >>> zeo.notifyConnected(ZEO.tests.servertesting.Connection())
>>> zeo.register('1', False)
- The ZEOStorage should sort out the following call such that the storage gets
- the correct parameters and so should return the parameters it was called with:
+ The ZEOStorage should sort out the following call such that the
+ storage gets the correct parameters and so should return the
+ parameters it was called with:
>>> zeo.history('oid',99)
('oid', '', 99)
@@ -181,7 +201,7 @@
>>> from ZEO.StorageServer import ZEOStorage308Adapter
>>> zeo = ZEOStorage308Adapter(VersionCompatibleStorage())
-
+
The history method should still return the parameters it was called with:
>>> zeo.history('oid','',99)
Modified: ZODB/branches/jim-thready/src/ZEO/tests/testZEO.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/tests/testZEO.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/tests/testZEO.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -25,7 +25,6 @@
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_unpickle
-import asyncore
import doctest
import logging
import os
@@ -464,7 +463,11 @@
pass
time.sleep(.1)
- self.failIf(self._storage.is_connected())
+ try:
+ self.failIf(self._storage.is_connected())
+ except:
+ print log
+ raise
self.assertEqual(len(ZEO.zrpc.connection.client_map), 1)
del ZEO.zrpc.connection.client_logger.critical
self.assertEqual(log[0][0], 'The ZEO client loop failed.')
@@ -738,6 +741,14 @@
blob_cache_dir = 'blobs'
shared_blob_dir = True
+class FauxConn:
+ addr = 'x'
+ peer_protocol_version = (
+ ZEO.zrpc.connection.Connection.current_protocol)
+
+ def auth_done(self):
+ pass
+
class StorageServerClientWrapper:
def __init__(self):
@@ -754,8 +765,8 @@
def __init__(self, server, storage_id):
self.storage_id = storage_id
self.server = ZEO.StorageServer.ZEOStorage(server, server.read_only)
+ self.server.notifyConnected(FauxConn())
self.server.register(storage_id, False)
- self.server._thunk = lambda : None
self.server.client = StorageServerClientWrapper()
def sortKey(self):
@@ -777,7 +788,6 @@
self.server.tpc_begin(id(transaction), '', '', {}, None, ' ')
def tpc_vote(self, transaction):
- self.server._restart()
self.server.vote(id(transaction))
result = self.server.client.serials[:]
del self.server.client.serials[:]
@@ -860,6 +870,8 @@
>>> fs = FileStorage('t.fs')
>>> sv = StorageServer(('', get_port()), dict(fs=fs))
>>> s = ZEOStorage(sv, sv.read_only)
+
+ >>> s.notifyConnected(FauxConn())
>>> s.register('fs', False)
If we ask for the last transaction, we should get the last transaction
Modified: ZODB/branches/jim-thready/src/ZEO/tests/testZEO2.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/tests/testZEO2.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/tests/testZEO2.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -92,9 +92,9 @@
handled correctly:
>>> zs1.tpc_abort('0') # doctest: +ELLIPSIS
+ (511/test-addr) ('1') unlock: transactions waiting: 0
2 callAsync serialnos ...
reply 1 None
- (511/test-addr) Blocked transaction restarted.
>>> fs.tpc_transaction() is not None
True
Modified: ZODB/branches/jim-thready/src/ZEO/zrpc/client.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/zrpc/client.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/zrpc/client.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -84,7 +84,6 @@
try:
t = self.thread
self.thread = None
- conn = self.connection
finally:
self.cond.release()
if t is not None:
@@ -94,9 +93,6 @@
if t.isAlive():
log("CM.close(): self.thread.join() timed out",
level=logging.WARNING)
- if conn is not None:
- # This will call close_conn() below which clears self.connection
- conn.close()
def attempt_connect(self):
"""Attempt a connection to the server without blocking too long.
Modified: ZODB/branches/jim-thready/src/ZEO/zrpc/connection.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/zrpc/connection.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/zrpc/connection.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -16,17 +16,16 @@
import errno
import select
import sys
+import thread
import threading
import logging
-import traceback, time
-
from ZEO.zrpc import smac
from ZEO.zrpc.error import ZRPCError, DisconnectedError
from ZEO.zrpc.marshal import Marshaller, ServerMarshaller
from ZEO.zrpc.trigger import trigger
from ZEO.zrpc.log import short_repr, log
-from ZODB.loglevels import BLATHER, TRACE
+from ZODB.loglevels import BLATHER
import ZODB.POSException
REPLY = ".reply" # message name used for replies
@@ -144,7 +143,7 @@
if obj is client_trigger:
continue
try:
- obj.mgr.client.close()
+ obj.mgr.client.close(True)
except:
map.pop(fd, None)
try:
@@ -188,7 +187,9 @@
def reply(self, obj):
self.send_reply(self.msgid, obj)
- def error(self, exc_info):
+ def error(self, exc_info=None):
+ if exc_info is None:
+ exc_info = sys.exc_info()
log("Error raised in delayed method", logging.ERROR, exc_info=True)
self.return_error(self.msgid, 0, *exc_info[:2])
@@ -360,15 +361,17 @@
# Protocol variables:
# Our preferred protocol.
- current_protocol = "Z309"
+ current_protocol = "Z310"
# If we're a client, an exhaustive list of the server protocols we
# can accept.
- servers_we_can_talk_to = ["Z308", current_protocol]
+ servers_we_can_talk_to = ["Z308", "Z309", current_protocol]
# If we're a server, an exhaustive list of the client protocols we
# can accept.
- clients_we_can_talk_to = ["Z200", "Z201", "Z303", "Z308", current_protocol]
+ clients_we_can_talk_to = [
+ "Z200", "Z201", "Z303", "Z308", "Z309",
+ current_protocol]
# This is pretty excruciating. Details:
#
@@ -759,11 +762,6 @@
self.log("wait(%d)" % msgid, level=TRACE)
self.trigger.pull_trigger()
-
- # Delay used when we call asyncore.poll() directly.
- # Start with a 1 msec delay, double until 1 sec.
- delay = 0.001
-
self.replies_cond.acquire()
try:
while 1:
@@ -794,7 +792,6 @@
self.trigger.pull_trigger()
-
class ManagedServerConnection(Connection):
"""Server-side Connection subclass."""
@@ -802,13 +799,19 @@
unlogged_exception_types = (ZODB.POSException.POSKeyError, )
# Servers use a shared server trigger that uses the asyncore socket map
- trigger = trigger()
+ #trigger = trigger()
def __init__(self, sock, addr, obj, mgr):
self.mgr = mgr
- Connection.__init__(self, sock, addr, obj, 'S')
+ map={}
+ Connection.__init__(self, sock, addr, obj, 'S', map=map)
self.marshal = ServerMarshaller()
+ self.trigger = trigger(map)
+ t = threading.Thread(target=server_loop, args=(map, self))
+ t.setDaemon(True)
+ t.start()
+
def handshake(self):
# Send the server's preferred protocol to the client.
self.message_output(self.current_protocol)
@@ -821,6 +824,27 @@
self.obj.notifyDisconnected()
Connection.close(self)
+ thread_ident = unregistered_thread_ident = None
+ def poll(self):
+ "Invoke asyncore mainloop to get pending message out."
+ ident = self.thread_ident
+ if ident is not None and thread.get_ident() == ident:
+ self.handle_write()
+ else:
+ self.trigger.pull_trigger()
+
+ def auth_done(self):
+ # We're done with the auth dance. We can be fast now.
+ self.thread_ident = self.unregistered_thread_ident
+
+def server_loop(map, conn):
+ conn.unregistered_thread_ident = thread.get_ident()
+
+ while len(map) > 1:
+ asyncore.poll(30.0, map)
+ for o in map.values():
+ o.close()
+
class ManagedClientConnection(Connection):
"""Client-side Connection subclass."""
__super_init = Connection.__init__
Modified: ZODB/branches/jim-thready/src/ZEO/zrpc/trigger.py
===================================================================
--- ZODB/branches/jim-thready/src/ZEO/zrpc/trigger.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZEO/zrpc/trigger.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -12,6 +12,8 @@
#
##############################################################################
+from __future__ import with_statement
+
import asyncore
import os
import socket
@@ -95,14 +97,15 @@
def _close(self): # see close() above; subclass must supply
raise NotImplementedError
- def pull_trigger(self, thunk=None):
+ def pull_trigger(self, *thunk):
if thunk:
- self.lock.acquire()
- try:
+ with self.lock:
self.thunks.append(thunk)
- finally:
- self.lock.release()
- self._physical_pull()
+ try:
+ self._physical_pull()
+ except Exception:
+ if not self._closed:
+ raise
# Subclass must supply _physical_pull, which does whatever the OS
# needs to do to provoke the "write" end of the trigger.
@@ -114,19 +117,20 @@
self.recv(8192)
except socket.error:
return
- self.lock.acquire()
- try:
- for thunk in self.thunks:
- try:
- thunk()
- except:
- nil, t, v, tbinfo = asyncore.compact_traceback()
- print ('exception in trigger thunk:'
- ' (%s:%s %s)' % (t, v, tbinfo))
- self.thunks = []
- finally:
- self.lock.release()
+ while 1:
+ with self.lock:
+ if self.thunks:
+ thunk = self.thunks.pop(0)
+ else:
+ return
+ try:
+ thunk[0](*thunk[1:])
+ except:
+ nil, t, v, tbinfo = asyncore.compact_traceback()
+ print ('exception in trigger thunk:'
+ ' (%s:%s %s)' % (t, v, tbinfo))
+
def __repr__(self):
return '<select-trigger (%s) at %x>' % (self.kind, positive_id(self))
Modified: ZODB/branches/jim-thready/src/ZODB/Connection.py
===================================================================
--- ZODB/branches/jim-thready/src/ZODB/Connection.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZODB/Connection.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -666,32 +666,11 @@
self._cache.update_object_size_estimation(oid, len(p))
obj._p_estimated_size = len(p)
- self._handle_serial(s, oid)
+ self._handle_serial(oid, s)
- def _handle_serial(self, store_return, oid=None, change=1):
- """Handle the returns from store() and tpc_vote() calls."""
-
- # These calls can return different types depending on whether
- # ZEO is used. ZEO uses asynchronous returns that may be
- # returned in batches by the ClientStorage. ZEO1 can also
- # return an exception object and expect that the Connection
- # will raise the exception.
-
- # When conflict resolution occurs, the object state held by
- # the connection does not match what is written to the
- # database. Invalidate the object here to guarantee that
- # the new state is read the next time the object is used.
-
- if not store_return:
+ def _handle_serial(self, oid, serial, change=True):
+ if not serial:
return
- if isinstance(store_return, str):
- assert oid is not None
- self._handle_one_serial(oid, store_return, change)
- else:
- for oid, serial in store_return:
- self._handle_one_serial(oid, serial, change)
-
- def _handle_one_serial(self, oid, serial, change):
if not isinstance(serial, str):
raise serial
obj = self._cache.get(oid, None)
@@ -757,7 +736,9 @@
except AttributeError:
return
s = vote(transaction)
- self._handle_serial(s)
+ if s:
+ for oid, serial in s:
+ self._handle_serial(oid, serial)
def tpc_finish(self, transaction):
"""Indicate confirmation that the transaction is done."""
@@ -1171,7 +1152,7 @@
s = self._storage.store(oid, serial, data,
'', transaction)
- self._handle_serial(s, oid, change=False)
+ self._handle_serial(oid, s, change=False)
src.close()
def _abort_savepoint(self):
Modified: ZODB/branches/jim-thready/src/ZODB/FileStorage/FileStorage.py
===================================================================
--- ZODB/branches/jim-thready/src/ZODB/FileStorage/FileStorage.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZODB/FileStorage/FileStorage.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -36,6 +36,7 @@
import logging
import os
import sys
+import threading
import time
import ZODB.blob
import ZODB.interfaces
@@ -128,7 +129,7 @@
else:
self._tfile = None
- self._file_name = file_name
+ self._file_name = os.path.abspath(file_name)
self._pack_gc = pack_gc
self.pack_keep_old = pack_keep_old
@@ -167,6 +168,7 @@
self._file = open(file_name, 'w+b')
self._file.write(packed_version)
+ self._files = FilePool(self._file_name)
r = self._restore_index()
if r is not None:
self._used_index = 1 # Marker for testing
@@ -401,6 +403,7 @@
def close(self):
self._file.close()
+ self._files.close()
if hasattr(self,'_lock_file'):
self._lock_file.close()
if self._tfile:
@@ -426,22 +429,25 @@
"""Return pickle data and serial number."""
assert not version
- self._lock_acquire()
+ _file = self._files.get()
try:
+
pos = self._lookup_pos(oid)
- h = self._read_data_header(pos, oid)
+
+ h = self._read_data_header(pos, oid, _file)
if h.plen:
- data = self._file.read(h.plen)
+ data = _file.read(h.plen)
return data, h.tid
elif h.back:
# Get the data from the backpointer, but tid from
# current txn.
- data = self._loadBack_impl(oid, h.back)[0]
+ data = self._loadBack_impl(oid, h.back, _file=_file)[0]
return data, h.tid
else:
raise POSKeyError(oid)
+
finally:
- self._lock_release()
+ self._files.put(_file)
def loadSerial(self, oid, serial):
self._lock_acquire()
@@ -462,12 +468,13 @@
self._lock_release()
def loadBefore(self, oid, tid):
- self._lock_acquire()
+ _file = self._files.get()
try:
pos = self._lookup_pos(oid)
+
end_tid = None
while True:
- h = self._read_data_header(pos, oid)
+ h = self._read_data_header(pos, oid, _file)
if h.tid < tid:
break
@@ -477,14 +484,14 @@
return None
if h.back:
- data, _, _, _ = self._loadBack_impl(oid, h.back)
+ data, _, _, _ = self._loadBack_impl(oid, h.back, _file=_file)
return data, h.tid, end_tid
else:
- return self._file.read(h.plen), h.tid, end_tid
-
+ return _file.read(h.plen), h.tid, end_tid
finally:
- self._lock_release()
+ self._files.put(_file)
+
def store(self, oid, oldserial, data, version, transaction):
if self._is_read_only:
raise POSException.ReadOnlyError()
@@ -735,6 +742,31 @@
finally:
self._lock_release()
+ def tpc_finish(self, transaction, f=None):
+
+ # Get write lock
+ self._files.write_lock()
+ try:
+ self._lock_acquire()
+ try:
+ if transaction is not self._transaction:
+ return
+ try:
+ if f is not None:
+ f(self._tid)
+ u, d, e = self._ude
+ self._finish(self._tid, u, d, e)
+ self._clear_temp()
+ finally:
+ self._ude = None
+ self._transaction = None
+ self._commit_lock_release()
+ finally:
+ self._lock_release()
+
+ finally:
+ self._files.write_unlock()
+
def _finish(self, tid, u, d, e):
# If self._nextpos is 0, then the transaction didn't write any
# data, so we don't bother writing anything to the file.
@@ -1131,8 +1163,10 @@
return
have_commit_lock = True
opos, index = pack_result
+ self._files.write_lock()
self._lock_acquire()
try:
+ self._files.empty()
self._file.close()
try:
os.rename(self._file_name, oldpath)
@@ -1146,6 +1180,7 @@
self._initIndex(index, self._tindex)
self._pos = opos
finally:
+ self._files.write_unlock()
self._lock_release()
# We're basically done. Now we need to deal with removed
@@ -2037,3 +2072,72 @@
'description': d}
d.update(e)
return d
+
+class FilePool:
+
+ closed = False
+ writing = False
+
+ def __init__(self, file_name):
+ self.name = file_name
+ self._files = []
+ self._out = []
+ self._cond = threading.Condition()
+
+ def write_lock(self):
+ self._cond.acquire()
+ try:
+ self.writing = True
+ while self._out:
+ self._cond.wait()
+ finally:
+ self._cond.release()
+
+ def write_unlock(self):
+ self._cond.acquire()
+ self.writing = False
+ self._cond.notifyAll()
+ self._cond.release()
+
+ def get(self):
+ self._cond.acquire()
+ try:
+ while self.writing:
+ self._cond.wait()
+ if self.closed:
+ raise ValueError('closed')
+
+ try:
+ f = self._files.pop()
+ except IndexError:
+ f = open(self.name, 'rb')
+ self._out.append(f)
+ return f
+ finally:
+ self._cond.release()
+
+ def put(self, f):
+ self._out.remove(f)
+ self._files.append(f)
+ if not self._out:
+ self._cond.acquire()
+ try:
+ if self.writing and not self._out:
+ self._cond.notifyAll()
+ finally:
+ self._cond.release()
+
+ def empty(self):
+ while self._files:
+ self._files.pop().close()
+
+ def close(self):
+ self._cond.acquire()
+ self.closed = True
+ self._cond.release()
+
+ self.write_lock()
+ try:
+ self.empty()
+ finally:
+ self.write_unlock()
Modified: ZODB/branches/jim-thready/src/ZODB/FileStorage/format.py
===================================================================
--- ZODB/branches/jim-thready/src/ZODB/FileStorage/format.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZODB/FileStorage/format.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -134,21 +134,24 @@
self._file.seek(pos)
return u64(self._file.read(8))
- def _read_data_header(self, pos, oid=None):
+ def _read_data_header(self, pos, oid=None, _file=None):
"""Return a DataHeader object for data record at pos.
If ois is not None, raise CorruptedDataError if oid passed
does not match oid in file.
"""
- self._file.seek(pos)
- s = self._file.read(DATA_HDR_LEN)
+ if _file is None:
+ _file = self._file
+
+ _file.seek(pos)
+ s = _file.read(DATA_HDR_LEN)
if len(s) != DATA_HDR_LEN:
raise CorruptedDataError(oid, s, pos)
h = DataHeaderFromString(s)
if oid is not None and oid != h.oid:
raise CorruptedDataError(oid, s, pos)
if not h.plen:
- h.back = u64(self._file.read(8))
+ h.back = u64(_file.read(8))
return h
def _read_txn_header(self, pos, tid=None):
@@ -164,20 +167,22 @@
h.ext = self._file.read(h.elen)
return h
- def _loadBack_impl(self, oid, back, fail=True):
+ def _loadBack_impl(self, oid, back, fail=True, _file=None):
# shared implementation used by various _loadBack methods
#
# If the backpointer ultimately resolves to 0:
# If fail is True, raise KeyError for zero backpointer.
# If fail is False, return the empty data from the record
# with no backpointer.
+ if _file is None:
+ _file = self._file
while 1:
if not back:
# If backpointer is 0, object does not currently exist.
raise POSKeyError(oid)
- h = self._read_data_header(back)
+ h = self._read_data_header(back, _file=_file)
if h.plen:
- return self._file.read(h.plen), h.tid, back, h.tloc
+ return _file.read(h.plen), h.tid, back, h.tloc
if h.back == 0 and not fail:
return None, h.tid, back, h.tloc
back = h.back
Modified: ZODB/branches/jim-thready/src/ZODB/tests/RevisionStorage.py
===================================================================
--- ZODB/branches/jim-thready/src/ZODB/tests/RevisionStorage.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZODB/tests/RevisionStorage.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -122,7 +122,7 @@
tid = info[0]["id"]
# Always undo the most recent txn, so the value will
# alternate between 3 and 4.
- self._undo(tid, [oid], note="undo %d" % i)
+ self._undo(tid, note="undo %d" % i)
revs.append(self._storage.load(oid, ""))
prev_tid = None
Modified: ZODB/branches/jim-thready/src/ZODB/tests/StorageTestBase.py
===================================================================
--- ZODB/branches/jim-thready/src/ZODB/tests/StorageTestBase.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZODB/tests/StorageTestBase.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -209,10 +209,11 @@
t = transaction.Transaction()
t.note(note or "undo")
self._storage.tpc_begin(t)
- tid, oids = self._storage.undo(tid, t)
+ undo_result = self._storage.undo(tid, t)
self._storage.tpc_vote(t)
self._storage.tpc_finish(t)
if expected_oids is not None:
+ oids = undo_result[1]
self.assertEqual(len(oids), len(expected_oids), repr(oids))
for oid in expected_oids:
self.assert_(oid in oids)
Modified: ZODB/branches/jim-thready/src/ZODB/tests/testFileStorage.py
===================================================================
--- ZODB/branches/jim-thready/src/ZODB/tests/testFileStorage.py 2010-01-15 19:47:08 UTC (rev 108161)
+++ ZODB/branches/jim-thready/src/ZODB/tests/testFileStorage.py 2010-01-15 19:47:43 UTC (rev 108162)
@@ -587,10 +587,10 @@
>>> handler.uninstall()
- >>> fs.load('\0'*8, '')
+ >>> fs.load('\0'*8, '') # doctest: +ELLIPSIS
Traceback (most recent call last):
...
- ValueError: I/O operation on closed file
+ ValueError: ...
>>> db.close()
>>> fs = ZODB.FileStorage.FileStorage('data.fs')