Commit 8b4689d5 authored by Jim Fulton's avatar Jim Fulton

provide IMultiCommitStorage

parent 2529c25b
......@@ -53,10 +53,7 @@ logger = logging.getLogger(__name__)
# max signed 64-bit value ~ infinity :) Signed cuz LBTree and TimeStamp
m64 = b'\x7f\xff\xff\xff\xff\xff\xff\xff'
try:
from ZODB.ConflictResolution import ResolvedSerial
except ImportError:
ResolvedSerial = 'rs'
from ZODB.ConflictResolution import ResolvedSerial
def tid2time(tid):
return str(TimeStamp(tid))
......@@ -77,6 +74,7 @@ def get_timestamp(prev_ts=None):
MB = 1024**2
@zope.interface.implementer(ZODB.interfaces.IMultiCommitStorage)
class ClientStorage(object):
"""A storage class that is a network client to a remote storage.
......@@ -726,7 +724,8 @@ class ClientStorage(object):
"""
tbuf = self._check_trans(txn, 'tpc_vote')
try:
self._call('vote', id(txn))
for oid in self._call('vote', id(txn)) or ():
tbuf.serial(oid, ResolvedSerial)
except POSException.ConflictError as err:
oid = getattr(err, 'oid', None)
if oid is not None:
......@@ -743,8 +742,8 @@ class ClientStorage(object):
if tbuf.exception:
raise tbuf.exception
if tbuf.serials:
return list(tbuf.serials.items())
if tbuf.resolved:
return list(tbuf.resolved)
else:
return None
......@@ -839,6 +838,8 @@ class ClientStorage(object):
self._update_blob_cache(tbuf, tid)
return tid
def _update_blob_cache(self, tbuf, tid):
"""Internal helper move blobs updated by a transaction to the cache.
"""
......
......@@ -427,7 +427,6 @@ class ZEOStorage:
for op, args in self.txnlog:
getattr(self, op)(*args)
# Blob support
while self.blob_log:
oid, oldserial, data, blobfilename = self.blob_log.pop()
......@@ -435,9 +434,11 @@ class ZEOStorage:
serials = self.storage.tpc_vote(self.transaction)
if serials:
self.serials.extend(serials)
if not isinstance(serials[0], bytes):
serials = (oid for (oid, serial) in serials
if serial == ResolvedSerial)
self.connection.async('serialnos', self.serials)
self.serials.extend(serials)
except Exception as err:
self.storage.tpc_abort(self.transaction)
......@@ -455,9 +456,9 @@ class ZEOStorage:
raise
else:
if delay is not None:
delay.reply(None)
delay.reply(self.serials)
else:
return None
return self.serials
else:
return delay
......@@ -567,17 +568,18 @@ class ZEOStorage:
if serial != b"\0\0\0\0\0\0\0\0":
self.invalidated.append(oid)
if newserial:
if isinstance(newserial, bytes):
newserial = [(oid, newserial)]
for oid, s in newserial or ():
for oid, s in newserial:
if s == ResolvedSerial:
self.stats.conflicts_resolved += 1
self.log("conflict resolved oid=%s"
% oid_repr(oid), BLATHER)
self.serials.append((oid, s))
self.serials.append(oid)
def _restore(self, oid, serial, data, prev_txn):
self.storage.restore(oid, serial, data, '', prev_txn,
......@@ -586,7 +588,7 @@ class ZEOStorage:
def _undo(self, trans_id):
tid, oids = self.storage.undo(trans_id, self.transaction)
self.invalidated.extend(oids)
self.serials.extend((oid, ResolvedSerial) for oid in oids)
self.serials.extend(oids)
# IStorageIteration support
......
......@@ -46,7 +46,7 @@ class TransactionBuffer:
# stored are builtin types -- strings or None.
self.pickler = Pickler(self.file, 1)
self.pickler.fast = 1
self.serials = {} # processed { oid -> serial }
self.resolved = set() # {oid}
self.exception = None
def close(self):
......@@ -61,10 +61,9 @@ class TransactionBuffer:
def serial(self, oid, serial):
if isinstance(serial, Exception):
self.exception = serial
self.serials[oid] = None
else:
self.serials[oid] = serial
self.exception = serial # This transaction will never be committed
elif serial == ResolvedSerial:
self.resolved.add(oid)
def storeBlob(self, oid, blobfilename):
self.blobs.append((oid, blobfilename))
......@@ -72,7 +71,7 @@ class TransactionBuffer:
def __iter__(self):
self.file.seek(0)
unpickler = Unpickler(self.file)
serials = self.serials
resolved = self.resolved
# Gaaaa, this is awkward. There can be entries in serials that
# aren't in the buffer, because undo. Entries can be repeated
......@@ -83,9 +82,9 @@ class TransactionBuffer:
for i in range(self.count):
oid, data = unpickler.load()
seen.add(oid)
yield oid, data, serials.get(oid) == ResolvedSerial
yield oid, data, oid in resolved
# We may have leftover serials because undo
for oid, serial in serials.items():
# We may have leftover oids because undo
for oid in resolved:
if oid not in seen:
yield oid, None, serial == ResolvedSerial
yield oid, None, True
......@@ -731,24 +731,23 @@ class StorageServerWrapper:
self.server.tpc_begin(id(transaction), '', '', {}, None, ' ')
def tpc_vote(self, transaction):
vote_result = self.server.vote(id(transaction))
assert vote_result is None
result = self.server.connection.serials[:]
result = self.server.vote(id(transaction))
assert result == self.server.connection.serials[:]
del self.server.connection.serials[:]
return result
def store(self, oid, serial, data, version_ignored, transaction):
self.server.storea(oid, serial, data, id(transaction))
def send_reply(self, *args): # Masquerade as conn
pass
def send_reply(self, _, result): # Masquerade as conn
self._result = result
def tpc_abort(self, transaction):
self.server.tpc_abort(id(transaction))
def tpc_finish(self, transaction, func = lambda: None):
self.server.tpc_finish(id(transaction)).set_sender(0, self)
return self._result
def multiple_storages_invalidation_queue_is_not_insane():
"""
......@@ -915,14 +914,14 @@ def tpc_finish_error():
buffer, sadly, using implementation details:
>>> tbuf = t.data(client)
>>> tbuf.serials = None
>>> tbuf.resolved = None
tpc_finish will fail:
>>> client.tpc_finish(t) # doctest: +ELLIPSIS
Traceback (most recent call last):
...
AttributeError: ...
TypeError: ...
>>> client.tpc_abort(t)
>>> t.abort()
......
......@@ -229,7 +229,7 @@ We start a transaction and vote, this leads to getting the lock.
ZEO.asyncio.server INFO
received handshake b'Z5'
>>> tid1 = start_trans(zs1)
>>> zs1.vote(tid1) # doctest: +ELLIPSIS
>>> resolved1 = zs1.vote(tid1) # doctest: +ELLIPSIS
ZEO.StorageServer DEBUG
(test-addr-1) ('1') lock: transactions waiting: 0
ZEO.StorageServer BLATHER
......@@ -486,7 +486,7 @@ ZEOStorage as closed and see if trying to get a lock cleans it up:
ZEO.asyncio.server INFO
received handshake b'Z5'
>>> tid1 = start_trans(zs1)
>>> zs1.vote(tid1) # doctest: +ELLIPSIS
>>> resolved1 = zs1.vote(tid1) # doctest: +ELLIPSIS
ZEO.StorageServer DEBUG
(test-addr-1) ('1') lock: transactions waiting: 0
ZEO.StorageServer BLATHER
......@@ -502,7 +502,7 @@ ZEOStorage as closed and see if trying to get a lock cleans it up:
ZEO.asyncio.server INFO
received handshake b'Z5'
>>> tid2 = start_trans(zs2)
>>> zs2.vote(tid2) # doctest: +ELLIPSIS
>>> resolved2 = zs2.vote(tid2) # doctest: +ELLIPSIS
ZEO.StorageServer DEBUG
(test-addr-2) ('1') lock: transactions waiting: 0
ZEO.StorageServer BLATHER
......
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