Commit d90c5b83 by Julien Muchembled

Allow NEO to store empty values

This changes how NEO stores undo information
and how it is transmitted on the network.
1 parent 03bf302a
Change History
==============
0.10 (unreleased)
-----------------
- NEO learned to store empty values (although it's useless when managed by
a ZODB Connection).
0.9.2 (unreleased)
------------------
......
......@@ -64,6 +64,9 @@ else:
compress = real_compress
makeChecksum = real_makeChecksum
CHECKED_SERIAL = object()
class Application(object):
"""The client node application."""
......@@ -427,15 +430,12 @@ class Application(object):
self._cache.store(oid, *result)
finally:
self._cache_lock_release()
if result[0] == '':
raise NEOStorageCreationUndoneError(dump(oid))
return result
finally:
self._load_lock_release()
@profiler_decorator
def _loadFromStorage(self, oid, at_tid, before_tid):
data = None
packet = Packets.AskObject(oid, at_tid, before_tid)
for node, conn in self.cp.iterateForObject(oid, readable=True):
try:
......@@ -444,23 +444,18 @@ class Application(object):
except ConnectionClosed:
continue
if checksum != makeChecksum(data):
# Warning: see TODO file.
# Check checksum.
neo.lib.logging.error('wrong checksum from %s for oid %s',
if data or checksum:
if checksum != makeChecksum(data):
neo.lib.logging.error('wrong checksum from %s for oid %s',
conn, dump(oid))
data = None
continue
break
if data is None:
# We didn't got any object from all storage node because of
# connection error
raise NEOStorageError('connection failure')
# Uncompress data
if compression:
data = decompress(data)
return data, tid, next_tid
continue
if compression:
data = decompress(data)
return data, tid, next_tid
raise NEOStorageCreationUndoneError(dump(oid))
# We didn't got any object from all storage node because of
# connection error
raise NEOStorageError('connection failure')
@profiler_decorator
def _loadFromCache(self, oid, at_tid=None, before_tid=None):
......@@ -512,8 +507,9 @@ class Application(object):
# This is some undo: either a no-data object (undoing object
# creation) or a back-pointer to an earlier revision (going back to
# an older object revision).
data = compressed_data = ''
compressed_data = ''
compression = 0
checksum = 0
else:
assert data_serial is None
compression = self.compress
......@@ -525,7 +521,7 @@ class Application(object):
compression = 0
else:
compression = 1
checksum = makeChecksum(compressed_data)
checksum = makeChecksum(compressed_data)
on_timeout = OnTimeout(self.onStoreTimeout, txn_context, oid)
# Store object in tmp cache
data_dict = txn_context['data_dict']
......@@ -600,11 +596,11 @@ class Application(object):
dump(oid), dump(serial))
for store_oid, store_data in data_dict.iteritems():
store_serial = object_serial_dict[store_oid]
if store_data is None:
if store_data is CHECKED_SERIAL:
self._checkCurrentSerialInTransaction(txn_context,
store_oid, store_serial)
else:
if store_data is '':
if store_data is None:
# Some undo
neo.lib.logging.warning('Deadlock avoidance cannot'
' reliably work with undo, this must be '
......@@ -615,7 +611,7 @@ class Application(object):
store_data, unlock=True)
else:
continue
elif data is not None:
elif data is not CHECKED_SERIAL:
resolved_serial_set = resolved_conflict_serial_dict.setdefault(
oid, set())
if resolved_serial_set and conflict_serial <= max(
......@@ -644,7 +640,7 @@ class Application(object):
# XXX: Is it really required to remove from data_dict ?
del data_dict[oid]
txn_context['data_list'].remove(oid)
if data is None:
if data is CHECKED_SERIAL:
raise ReadConflictError(oid=oid, serials=(conflict_serial,
serial))
raise ConflictError(oid=oid, serials=(txn_context['ttid'],
......@@ -789,14 +785,14 @@ class Application(object):
try:
cache = self._cache
for oid, data in txn_context['data_dict'].iteritems():
if data is None:
if data is CHECKED_SERIAL:
# this is just a remain of
# checkCurrentSerialInTransaction call, ignore (no data
# was modified).
continue
# Update ex-latest value in cache
cache.invalidate(oid, tid)
if data:
if data is not None:
# Store in cache with no next_tid
cache.store(oid, data, tid, None)
finally:
......@@ -1097,7 +1093,7 @@ class Application(object):
data_dict = txn_context['data_dict']
if oid not in data_dict:
# Marker value so we don't try to resolve conflicts.
data_dict[oid] = None
data_dict[oid] = CHECKED_SERIAL
txn_context['data_list'].append(oid)
packet = Packets.AskCheckCurrentSerial(ttid, serial, oid)
for node, conn in self.cp.iterateForObject(oid, writable=True):
......
......@@ -25,6 +25,7 @@ import neo.lib
from hashlib import md5
from neo.storage.database import DatabaseManager
from neo.storage.database.manager import CreationUndone
from neo.lib.protocol import CellStates, ZERO_OID, ZERO_TID
from neo.lib import util
......@@ -112,9 +113,6 @@ def _noPrune(_):
prune = _prune
class CreationUndone(Exception):
pass
def iterObjSerials(obj):
for tserial in obj.values():
for serial in tserial.keys():
......@@ -658,7 +656,7 @@ class BTreeDatabaseManager(DatabaseManager):
# No entry before pack TID, nothing to pack on this object.
pass
else:
if tserial[max_serial][2] == '':
if tserial[max_serial][1] is None:
# Last version before/at pack TID is a creation undo, drop
# it too.
max_serial += 1
......
......@@ -262,10 +262,7 @@ class DatabaseManager(object):
_, compression, checksum, data = self._getObjectData(oid,
data_serial, serial)
except CreationUndone:
compression = 0
# XXX: this is the valid checksum for empty string
checksum = 1
data = ''
pass
data_serial = None
if serial is not None:
serial = p64(serial)
......
......@@ -809,13 +809,13 @@ class MySQLDatabaseManager(DatabaseManager):
for count, oid, max_serial in q('SELECT COUNT(*) - 1, oid, '
'MAX(serial) FROM obj_short WHERE serial <= %(tid)d '
'GROUP BY oid' % {'tid': tid}):
if q('SELECT LENGTH(value) FROM obj WHERE partition ='
if q('SELECT 1 FROM obj WHERE partition ='
'%(partition)s AND oid = %(oid)d AND '
'serial = %(max_serial)d' % {
'serial = %(max_serial)d AND checksum IS NULL' % {
'oid': oid,
'partition': getPartition(oid),
'max_serial': max_serial,
})[0][0] == 0:
}):
count += 1
max_serial += 1
if count:
......
......@@ -96,6 +96,9 @@ class BaseClientAndStorageOperationHandler(EventHandler):
serial, next_serial, compression, checksum, data, data_serial = o
neo.lib.logging.debug('oid = %s, serial = %s, next_serial = %s',
dump(oid), dump(serial), dump(next_serial))
if checksum is None:
checksum = 0
data = ''
p = Packets.AnswerObject(oid, serial, next_serial,
compression, checksum, data, data_serial)
conn.answer(p)
......
......@@ -17,7 +17,7 @@
import neo.lib
from neo.lib import protocol
from neo.lib.util import dump
from neo.lib.util import dump, makeChecksum
from neo.lib.protocol import Packets, LockState, Errors
from neo.storage.handlers import BaseClientAndStorageOperationHandler
from neo.storage.transactions import ConflictError, DelayedError
......@@ -88,11 +88,12 @@ class ClientOperationHandler(BaseClientAndStorageOperationHandler):
compression, checksum, data, data_serial, ttid, unlock):
# register the transaction
self.app.tm.register(conn.getUUID(), ttid)
if data_serial is not None:
assert data == '', repr(data)
# Change data to None here, to do it only once, even if store gets
# delayed.
data = None
if data or checksum:
# TODO: return an appropriate error packet
assert makeChecksum(data) == checksum
assert data_serial is None
else:
checksum = data = None
self._askStoreObject(conn, oid, serial, compression, checksum, data,
data_serial, ttid, unlock, time.time())
......
......@@ -18,6 +18,7 @@
import unittest
from mock import Mock, ReturnValues
from collections import deque
from neo.lib.util import makeChecksum
from neo.tests import NeoUnitTestBase
from neo.storage.app import Application
from neo.storage.transactions import ConflictError, DelayedError
......@@ -207,7 +208,8 @@ class StorageClientHandlerTests(NeoUnitTestBase):
def _getObject(self):
oid = self.getOID(0)
serial = self.getNextTID()
return (oid, serial, 1, '1', 'DATA')
data = 'DATA'
return (oid, serial, 1, makeChecksum(data), data)
def _checkStoreObjectCalled(self, *args):
calls = self.app.tm.mockGetNamedCalls('storeObject')
......@@ -237,10 +239,10 @@ class StorageClientHandlerTests(NeoUnitTestBase):
tid = self.getNextTID()
oid, serial, comp, checksum, data = self._getObject()
data_tid = self.getNextTID()
self.operation.askStoreObject(conn, oid, serial, comp, checksum,
self.operation.askStoreObject(conn, oid, serial, comp, 0,
'', data_tid, tid, False)
self._checkStoreObjectCalled(tid, serial, oid, comp,
checksum, None, data_tid, False)
None, None, data_tid, False)
pconflicting, poid, pserial = self.checkAnswerStoreObject(conn,
decode=True)
self.assertEqual(pconflicting, 0)
......
......@@ -17,6 +17,7 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
import threading
import transaction
from thread import get_ident
from persistent import Persistent
from neo.storage.transactions import TransactionManager, \
......@@ -37,6 +38,25 @@ class PCounterWithResolution(PCounter):
class Test(NEOThreadedTest):
def testBasicStore(self):
cluster = NEOCluster()
try:
cluster.start()
storage = cluster.getZODBStorage()
for data in 'foo', '':
oid = storage.new_oid()
txn = transaction.Transaction()
storage.tpc_begin(txn)
r1 = storage.store(oid, None, data, '', txn)
r2 = storage.tpc_vote(txn)
serial = storage.tpc_finish(txn)
self.assertEqual((data, serial), storage.load(oid, ''))
storage._cache.clear()
self.assertEqual((data, serial), storage.load(oid, ''))
self.assertEqual((data, serial), storage.load(oid, ''))
finally:
cluster.stop()
def testDelayedUnlockInformation(self):
except_list = []
def delayUnlockInformation(conn, packet):
......
Styling with Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!