Commit 3c21d591 authored by Jim Fulton's avatar Jim Fulton

Bug fixed:

- The undo implementation was incorrect in ways that could cause
  subtle missbehaviors.

API changes:

- The API for undoing multiple transactions has changed.  To undo
  multiple transactions in a single transaction, pass pass a list of
  transaction identifiers to a database's undo method. Calling a
  database's undo method multiple times in the same transaction now
  raises an exception.

- The storage API (IStorage) has been tightened. Now, storages should
  raise a StorageTransactionError when invalid transactions are passed
  to tpc_begin, tpc_vote, or tpc_finish.
parent af947463
......@@ -8,6 +8,16 @@
New Features
------------
- The API for undoing multiple transactions has changed. To undo
multiple transactions in a single transaction, pass pass a list of
transaction identifiers to a database's undo method. Calling a
database's undo method multiple times in the same transaction now
raises an exception.
- The storage API (IStorage) has been tightened. Now, storages should
raise a StorageTransactionError when invalid transactions are passed
to tpc_begin, tpc_vote, or tpc_finish.
- Broken objects now provide the IBroken interface.
Bugs Fixed
......@@ -43,6 +53,9 @@ Bugs Fixed
- C Header files weren't installed correctly.
- The undo implementation was incorrect in ways that could cause
subtle missbehaviors.
3.9.3 (2009-10-23)
==================
......
......@@ -1075,7 +1075,8 @@ class ClientStorage(object):
def tpc_vote(self, txn):
"""Storage API: vote on a transaction."""
if txn is not self._transaction:
return
raise POSException.StorageTransactionError(
"tpc_vote called with wrong transaction")
self._server.vote(id(txn))
return self._check_serials()
......@@ -1094,7 +1095,9 @@ class ClientStorage(object):
# must be ignored.
if self._transaction == txn:
self._tpc_cond.release()
return
raise POSException.StorageTransactionError(
"Duplicate tpc_begin calls for same transaction")
self._tpc_cond.wait(30)
self._transaction = txn
self._tpc_cond.release()
......@@ -1148,7 +1151,8 @@ class ClientStorage(object):
def tpc_finish(self, txn, f=None):
"""Storage API: finish a transaction."""
if txn is not self._transaction:
return
raise POSException.StorageTransactionError(
"tpc_finish called with wrong transaction")
self._load_lock.acquire()
try:
if self._midtxn_disconnect:
......
......@@ -1093,7 +1093,8 @@ class TimeoutTests(CommonSetupTearDown):
self.assertRaises(ConflictError, storage.tpc_vote, t)
# Even aborting won't help.
storage.tpc_abort(t)
storage.tpc_finish(t)
self.assertRaises(ZODB.POSException.StorageTransactionError,
storage.tpc_finish, t)
# Try again.
obj.value = 10
t = Transaction()
......@@ -1103,7 +1104,7 @@ class TimeoutTests(CommonSetupTearDown):
self.assertRaises(ConflictError, storage.tpc_vote, t)
# Abort this one and try a transaction that should succeed.
storage.tpc_abort(t)
storage.tpc_finish(t)
# Now do a store.
obj.value = 11
t = Transaction()
......
......@@ -221,7 +221,8 @@ class BaseStorage(UndoLogCompatible):
self._lock_acquire()
try:
if self._transaction is transaction:
return
raise POSException.StorageTransactionError(
"Duplicate tpc_begin calls for same transaction")
self._lock_release()
self._commit_lock_acquire()
self._lock_acquire()
......@@ -264,7 +265,8 @@ class BaseStorage(UndoLogCompatible):
self._lock_acquire()
try:
if transaction is not self._transaction:
return
raise POSException.StorageTransactionError(
"tpc_vote called with wrong transaction")
self._vote()
finally:
self._lock_release()
......@@ -284,7 +286,8 @@ class BaseStorage(UndoLogCompatible):
self._lock_acquire()
try:
if transaction is not self._transaction:
return
raise POSException.StorageTransactionError(
"tpc_finish called with wrong transaction")
try:
if f is not None:
f(self._tid)
......
......@@ -896,7 +896,7 @@ class DB(object):
finally:
self._r()
def undo(self, id, txn=None):
def undo(self, ids, txn=None):
"""Undo a transaction identified by id.
A transaction can be undone if all of the objects involved in
......@@ -909,13 +909,16 @@ class DB(object):
transaction id used by other methods; it is unique to undo().
:Parameters:
- `id`: a storage-specific transaction identifier
- `ids`: a sequence of storage-specific transaction identifiers
or a single transaction identifier
- `txn`: transaction context to use for undo().
By default, uses the current transaction.
"""
if txn is None:
txn = transaction.get()
txn.register(TransactionalUndo(self, id))
if isinstance(ids, basestring):
ids = [ids]
txn.join(TransactionalUndo(self, ids))
def transaction(self):
return ContextManager(self)
......@@ -943,60 +946,41 @@ class ContextManager:
resource_counter_lock = threading.Lock()
resource_counter = 0
class ResourceManager(object):
"""Transaction participation for an undo resource."""
class TransactionalUndo(object):
# XXX This implementation is broken. Subclasses invalidate oids
# in their commit calls. Invalidations should not be sent until
# tpc_finish is called. In fact, invalidations should be sent to
# the db *while* tpc_finish is being called on the storage.
def __init__(self, db):
def __init__(self, db, tids):
self._db = db
# Delegate the actual 2PC methods to the storage
self.tpc_vote = self._db.storage.tpc_vote
self.tpc_finish = self._db.storage.tpc_finish
self.tpc_abort = self._db.storage.tpc_abort
# Get a number from a simple thread-safe counter, then
# increment it, for the purpose of sorting ResourceManagers by
# creation order. This ensures that multiple ResourceManagers
# within a transaction commit in a predictable sequence.
resource_counter_lock.acquire()
try:
global resource_counter
self._count = resource_counter
resource_counter += 1
finally:
resource_counter_lock.release()
self._storage = db.storage
self._tids = tids
self._oids = set()
def sortKey(self):
return "%s:%016x" % (self._db.storage.sortKey(), self._count)
def tpc_begin(self, txn, sub=False):
if sub:
raise ValueError("doesn't support sub-transactions")
self._db.storage.tpc_begin(txn)
def abort(self, transaction):
pass
# The object registers itself with the txn manager, so the ob
# argument to the methods below is self.
def tpc_begin(self, transaction):
self._storage.tpc_begin(transaction)
def abort(self, obj, txn):
raise NotImplementedError
def commit(self, transaction):
for tid in self._tids:
result = self._storage.undo(tid, transaction)
if result:
self._oids.update(result[1])
def commit(self, obj, txn):
raise NotImplementedError
def tpc_vote(self, transaction):
for oid, _ in self._storage.tpc_vote(transaction) or ():
self._oids.add(oid)
class TransactionalUndo(ResourceManager):
def tpc_finish(self, transaction):
self._storage.tpc_finish(
transaction,
lambda tid: self._db.invalidate(tid, self._oids)
)
def __init__(self, db, tid):
super(TransactionalUndo, self).__init__(db)
self._tid = tid
def tpc_abort(self, transaction):
self._storage.tpc_abort(transaction)
def commit(self, ob, t):
# XXX see XXX in ResourceManager
tid, oids = self._db.storage.undo(self._tid, t)
self._db.invalidate(tid, dict.fromkeys(oids, 1))
def sortKey(self):
return "%s:%s" % (self._storage.sortKey(), id(self))
def connection(*args, **kw):
db = DB(*args, **kw)
......
......@@ -309,7 +309,8 @@ class DemoStorage(object):
def tpc_begin(self, transaction, *a, **k):
# The tid argument exists to support testing.
if transaction is self._transaction:
return
raise ZODB.POSException.StorageTransactionError(
"Duplicate tpc_begin calls for same transaction")
self._lock_release()
self._commit_lock.acquire()
self._lock_acquire()
......@@ -320,7 +321,8 @@ class DemoStorage(object):
@ZODB.utils.locked
def tpc_finish(self, transaction, func = lambda tid: None):
if (transaction is not self._transaction):
return
raise ZODB.POSException.StorageTransactionError(
"tpc_finish called with wrong transaction")
self._issued_oids.difference_update(self._stored_oids)
self._stored_oids = set()
self._transaction = None
......
......@@ -705,7 +705,8 @@ class FileStorage(
self._lock_acquire()
try:
if transaction is not self._transaction:
return
raise POSException.StorageTransactionError(
"tpc_vote called with wrong transaction")
dlen = self._tfile.tell()
if not dlen:
return # No data in this trans
......
......@@ -274,7 +274,8 @@ class MappingStorage(object):
def tpc_begin(self, transaction, tid=None):
# The tid argument exists to support testing.
if transaction is self._transaction:
return
raise ZODB.POSException.StorageTransactionError(
"Duplicate tpc_begin calls for same transaction")
self._lock_release()
self._commit_lock.acquire()
self._lock_acquire()
......@@ -292,7 +293,8 @@ class MappingStorage(object):
@ZODB.utils.locked(opened)
def tpc_finish(self, transaction, func = lambda tid: None):
if (transaction is not self._transaction):
return
raise ZODB.POSException.StorageTransactionError(
"tpc_finish called with wrong transaction")
tid = self._tid
func(tid)
......@@ -318,7 +320,9 @@ class MappingStorage(object):
# ZODB.interfaces.IStorage
def tpc_vote(self, transaction):
pass
if transaction is not self._transaction:
raise POSException.StorageTransactionError(
"tpc_vote called with wrong transaction")
class TransactionRecord:
......
......@@ -690,7 +690,7 @@ class IStorage(Interface):
"""Begin the two-phase commit process.
If storage is already participating in a two-phase commit
using the same transaction, the call is ignored.
using the same transaction, a StorageTransactionError is raised.
If the storage is already participating in a two-phase commit
using a different transaction, the call blocks until the
......@@ -702,9 +702,10 @@ class IStorage(Interface):
Changes must be made permanent at this point.
This call is ignored if the storage isn't participating in
two-phase commit or if it is committing a different
transaction. Failure of this method is extremely serious.
This call raises a StorageTransactionError if the storage
isn't participating in two-phase commit or if it is committing
a different transaction. Failure of this method is extremely
serious.
The second argument is a call-back function that must be
called while the storage transaction lock is held. It takes
......@@ -715,9 +716,9 @@ class IStorage(Interface):
def tpc_vote(transaction):
"""Provide a storage with an opportunity to veto a transaction
This call is ignored if the storage isn't participating in
two-phase commit or if it is commiting a different
transaction. Failure of this method is extremely serious.
This call raises a StorageTransactionError if the storage
isn't participating in two-phase commit or if it is commiting
a different transaction.
If a transaction can be committed by a storage, then the
method should return. If a transaction cannot be committed,
......
......@@ -34,8 +34,8 @@ class BasicStorage:
def checkBasics(self):
t = transaction.Transaction()
self._storage.tpc_begin(t)
# This should simply return
self._storage.tpc_begin(t)
self.assertRaises(POSException.StorageTransactionError,
self._storage.tpc_begin, t)
# Aborting is easy
self._storage.tpc_abort(t)
# Test a few expected exceptions when we're doing operations giving a
......
......@@ -99,19 +99,20 @@ class SynchronizedStorage:
def checkFinishNotCommitting(self):
t = Transaction()
self._storage.tpc_finish(t)
self.assertRaises(StorageTransactionError,
self._storage.tpc_finish, t)
self._storage.tpc_abort(t)
def checkFinishWrongTrans(self):
t = Transaction()
self._storage.tpc_begin(t)
self._storage.tpc_finish(Transaction())
self.assertRaises(StorageTransactionError,
self._storage.tpc_finish, Transaction())
self._storage.tpc_abort(t)
def checkBeginCommitting(self):
t = Transaction()
self._storage.tpc_begin(t)
self._storage.tpc_begin(t)
self._storage.tpc_abort(t)
# TODO: how to check undo?
......@@ -420,8 +420,8 @@ class ZODBTests(ZODB.tests.util.TestCase):
# performed yet.
transaction.begin()
log = self._db.undoLog()
for i in range(5):
self._db.undo(log[i]['id'])
self._db.undo([log[i]['id'] for i in range(5)])
transaction.get().note('undo states 1 through 5')
# Now attempt all those undo operations.
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment