Commit d14040ac authored by Kirill Smelkov's avatar Kirill Smelkov

Sync with NEO/py v1.12-13-gf2ea4be2 (oldproto branch)

* origin/old-proto:
  qa: skip broken ZODB test
  client: fix race with invalidations when starting a new transaction on ZODB 5
  Code clean-up, comment fixes
  master: fix crash in STARTING_BACKUP when connecting to an upstream secondary master
  mysql: workaround for MDEV-20693
  client: inline Application._loadFromCache
  client: replace global load lock by a per-oid one
  client: unindent code
  client: remove load lock in tpc_finish
  qa: check cache in testExternalInvalidation
  qa: comment testExternalInvalidation2
parents 8ea85c71 f2ea4be2
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
import heapq import heapq
import random import random
import time import time
from collections import defaultdict
try: try:
from ZODB._compat import dumps, loads, _protocol from ZODB._compat import dumps, loads, _protocol
except ImportError: except ImportError:
...@@ -79,7 +79,7 @@ class Application(ThreadedApplication): ...@@ -79,7 +79,7 @@ class Application(ThreadedApplication):
# no self-assigned NID, primary master will supply us one # no self-assigned NID, primary master will supply us one
self._cache = ClientCache() if cache_size is None else \ self._cache = ClientCache() if cache_size is None else \
ClientCache(max_size=cache_size) ClientCache(max_size=cache_size)
self._loading_oid = None self._loading = defaultdict(lambda: (Lock(), []))
self.new_oid_list = () self.new_oid_list = ()
self.last_oid = '\0' * 8 self.last_oid = '\0' * 8
self.storage_event_handler = storage.StorageEventHandler(self) self.storage_event_handler = storage.StorageEventHandler(self)
...@@ -90,19 +90,13 @@ class Application(ThreadedApplication): ...@@ -90,19 +90,13 @@ class Application(ThreadedApplication):
self.notifications_handler = master.PrimaryNotificationsHandler( self) self.notifications_handler = master.PrimaryNotificationsHandler( self)
self._txn_container = TransactionContainer() self._txn_container = TransactionContainer()
# Lock definition : # Lock definition :
# _load_lock is used to make loading and storing atomic
lock = Lock()
self._load_lock_acquire = lock.acquire
self._load_lock_release = lock.release
# _oid_lock is used in order to not call multiple oid # _oid_lock is used in order to not call multiple oid
# generation at the same time # generation at the same time
lock = Lock() lock = Lock()
self._oid_lock_acquire = lock.acquire self._oid_lock_acquire = lock.acquire
self._oid_lock_release = lock.release self._oid_lock_release = lock.release
lock = Lock()
# _cache_lock is used for the client cache # _cache_lock is used for the client cache
self._cache_lock_acquire = lock.acquire self._cache_lock = Lock()
self._cache_lock_release = lock.release
# _connecting_to_master_node is used to prevent simultaneous master # _connecting_to_master_node is used to prevent simultaneous master
# node connection attempts # node connection attempts
self._connecting_to_master_node = Lock() self._connecting_to_master_node = Lock()
...@@ -397,21 +391,32 @@ class Application(ThreadedApplication): ...@@ -397,21 +391,32 @@ class Application(ThreadedApplication):
""" """
# TODO: # TODO:
# - rename parameters (here? and in handlers & packet definitions) # - rename parameters (here? and in handlers & packet definitions)
acquired = False
acquire = self._cache_lock_acquire lock = self._cache_lock
release = self._cache_lock_release
# XXX: Consider using a more fine-grained lock.
self._load_lock_acquire()
try:
acquire()
try: try:
result = self._loadFromCache(oid, tid, before_tid) while 1:
with lock:
if tid:
result = self._cache.load(oid, tid + '*')
assert not result or result[1] == tid
else:
result = self._cache.load(oid, before_tid)
if result: if result:
return result return result
self._loading_oid = oid load_lock = self._loading[oid][0]
self._loading_invalidated = [] acquired = load_lock.acquire(0)
finally: # Several concurrent cache misses for the same oid are probably
release() # for the same tid so we use a per-oid lock to avoid asking the
# same data to the storage node.
if acquired:
# The first thread does load from storage,
# and fills cache with the response.
break
# The other threads wait for the first one to complete and
# loop, possibly resulting in a new cache miss if a different
# tid is actually wanted or if the data was too big.
with load_lock:
pass
# While the cache lock is released, an arbitrary number of # While the cache lock is released, an arbitrary number of
# invalidations may be processed, for this oid or not. And at this # invalidations may be processed, for this oid or not. And at this
# precise moment, if both tid and before_tid are None (which is # precise moment, if both tid and before_tid are None (which is
...@@ -427,20 +432,24 @@ class Application(ThreadedApplication): ...@@ -427,20 +432,24 @@ class Application(ThreadedApplication):
# we got from master. # we got from master.
before_tid = p64(u64(self.last_tid) + 1) before_tid = p64(u64(self.last_tid) + 1)
data, tid, next_tid, _ = self._loadFromStorage(oid, tid, before_tid) data, tid, next_tid, _ = self._loadFromStorage(oid, tid, before_tid)
acquire() with lock:
try: loading = self._loading.pop(oid, None)
if self._loading_oid: if loading:
assert loading[0] is load_lock
if not next_tid: if not next_tid:
for t in self._loading_invalidated: for t in loading[1]:
if tid < t: if tid < t:
next_tid = t next_tid = t
break break
self._cache.store(oid, data, tid, next_tid) self._cache.store(oid, data, tid, next_tid)
# Else, we just reconnected to the master. # Else, we just reconnected to the master.
finally: load_lock.release()
release() except:
finally: if acquired:
self._load_lock_release() with lock:
self._loading.pop(oid, None)
load_lock.release()
raise
return data, tid, next_tid return data, tid, next_tid
def _loadFromStorage(self, oid, at_tid, before_tid): def _loadFromStorage(self, oid, at_tid, before_tid):
...@@ -459,16 +468,6 @@ class Application(ThreadedApplication): ...@@ -459,16 +468,6 @@ class Application(ThreadedApplication):
Packets.AskObject(oid, at_tid, before_tid), Packets.AskObject(oid, at_tid, before_tid),
askStorage) askStorage)
def _loadFromCache(self, oid, at_tid=None, before_tid=None):
"""
Load from local cache, return None if not found.
"""
if at_tid:
result = self._cache.load(oid, at_tid + '*')
assert not result or result[1] == at_tid
return result
return self._cache.load(oid, before_tid)
def tpc_begin(self, storage, transaction, tid=None, status=' '): def tpc_begin(self, storage, transaction, tid=None, status=' '):
"""Begin a new transaction.""" """Begin a new transaction."""
# First get a transaction, only one is allowed at a time # First get a transaction, only one is allowed at a time
...@@ -670,7 +669,7 @@ class Application(ThreadedApplication): ...@@ -670,7 +669,7 @@ class Application(ThreadedApplication):
txn_context = self._txn_container.pop(transaction) txn_context = self._txn_container.pop(transaction)
if txn_context is None: if txn_context is None:
return return
# We want that the involved nodes abort a transaction after any # We want the involved nodes to abort a transaction after any
# other packet sent by the client for this transaction. IOW, if we # other packet sent by the client for this transaction. IOW, if we
# already have a connection with a storage node, potentially with # already have a connection with a storage node, potentially with
# a pending write, aborting only via the master may lead to a race # a pending write, aborting only via the master may lead to a race
...@@ -699,9 +698,8 @@ class Application(ThreadedApplication): ...@@ -699,9 +698,8 @@ class Application(ThreadedApplication):
txn_context.conn_dict)) txn_context.conn_dict))
except ConnectionClosed: except ConnectionClosed:
pass pass
# We don't need to flush queue, as it won't be reused by future # No need to flush queue, as it will be destroyed on return,
# transactions (deleted on next line & indexed by transaction object # along with txn_context.
# instance).
self.dispatcher.forget_queue(txn_context.queue, flush_queue=False) self.dispatcher.forget_queue(txn_context.queue, flush_queue=False)
def tpc_finish(self, transaction, f=None): def tpc_finish(self, transaction, f=None):
...@@ -724,10 +722,6 @@ class Application(ThreadedApplication): ...@@ -724,10 +722,6 @@ class Application(ThreadedApplication):
txn_container = self._txn_container txn_container = self._txn_container
if not txn_container.get(transaction).voted: if not txn_container.get(transaction).voted:
self.tpc_vote(transaction) self.tpc_vote(transaction)
checked_list = []
self._load_lock_acquire()
try:
# Call finish on master
txn_context = txn_container.pop(transaction) txn_context = txn_container.pop(transaction)
cache_dict = txn_context.cache_dict cache_dict = txn_context.cache_dict
checked_list = [oid for oid, data in cache_dict.iteritems() checked_list = [oid for oid, data in cache_dict.iteritems()
...@@ -744,8 +738,6 @@ class Application(ThreadedApplication): ...@@ -744,8 +738,6 @@ class Application(ThreadedApplication):
if not tid: if not tid:
raise raise
return tid return tid
finally:
self._load_lock_release()
def _getFinalTID(self, ttid): def _getFinalTID(self, ttid):
try: try:
...@@ -991,11 +983,8 @@ class Application(ThreadedApplication): ...@@ -991,11 +983,8 @@ class Application(ThreadedApplication):
# It should not be otherwise required (clients should be free to load # It should not be otherwise required (clients should be free to load
# old data as long as it is available in cache, event if it was pruned # old data as long as it is available in cache, event if it was pruned
# by a pack), so don't bother invalidating on other clients. # by a pack), so don't bother invalidating on other clients.
self._cache_lock_acquire() with self._cache_lock:
try:
self._cache.clear() self._cache.clear()
finally:
self._cache_lock_release()
def getLastTID(self, oid): def getLastTID(self, oid):
return self.load(oid)[1] return self.load(oid)[1]
......
# -*- coding: utf-8 -*-
# #
# Copyright (C) 2006-2019 Nexedi SA # Copyright (C) 2006-2019 Nexedi SA
# #
...@@ -45,8 +46,7 @@ class PrimaryNotificationsHandler(MTEventHandler): ...@@ -45,8 +46,7 @@ class PrimaryNotificationsHandler(MTEventHandler):
# Either we're connecting or we already know the last tid # Either we're connecting or we already know the last tid
# via invalidations. # via invalidations.
assert app.master_conn is None, app.master_conn assert app.master_conn is None, app.master_conn
app._cache_lock_acquire() with app._cache_lock:
try:
if app_last_tid < ltid: if app_last_tid < ltid:
app._cache.clear_current() app._cache.clear_current()
# In the past, we tried not to invalidate the # In the past, we tried not to invalidate the
...@@ -60,9 +60,7 @@ class PrimaryNotificationsHandler(MTEventHandler): ...@@ -60,9 +60,7 @@ class PrimaryNotificationsHandler(MTEventHandler):
app._cache.clear() app._cache.clear()
# Make sure a parallel load won't refill the cache # Make sure a parallel load won't refill the cache
# with garbage. # with garbage.
app._loading_oid = app._loading_invalidated = None app._loading.clear()
finally:
app._cache_lock_release()
db = app.getDB() db = app.getDB()
db is None or db.invalidateCache() db is None or db.invalidateCache()
app.last_tid = ltid app.last_tid = ltid
...@@ -70,21 +68,22 @@ class PrimaryNotificationsHandler(MTEventHandler): ...@@ -70,21 +68,22 @@ class PrimaryNotificationsHandler(MTEventHandler):
def answerTransactionFinished(self, conn, _, tid, callback, cache_dict): def answerTransactionFinished(self, conn, _, tid, callback, cache_dict):
app = self.app app = self.app
app.last_tid = tid
# Update cache
cache = app._cache cache = app._cache
app._cache_lock_acquire() invalidate = cache.invalidate
try: loading_get = app._loading.get
with app._cache_lock:
for oid, data in cache_dict.iteritems(): for oid, data in cache_dict.iteritems():
# Update ex-latest value in cache # Update ex-latest value in cache
cache.invalidate(oid, tid) invalidate(oid, tid)
loading = loading_get(oid)
if loading:
loading[1].append(tid)
if data is not None: if data is not None:
# Store in cache with no next_tid # Store in cache with no next_tid
cache.store(oid, data, tid, None) cache.store(oid, data, tid, None)
if callback is not None: if callback is not None:
callback(tid) callback(tid)
finally: app.last_tid = tid # see comment in invalidateObjects
app._cache_lock_release()
def connectionClosed(self, conn): def connectionClosed(self, conn):
app = self.app app = self.app
...@@ -112,20 +111,24 @@ class PrimaryNotificationsHandler(MTEventHandler): ...@@ -112,20 +111,24 @@ class PrimaryNotificationsHandler(MTEventHandler):
app = self.app app = self.app
if app.ignore_invalidations: if app.ignore_invalidations:
return return
app.last_tid = tid with app._cache_lock:
app._cache_lock_acquire()
try:
invalidate = app._cache.invalidate invalidate = app._cache.invalidate
loading = app._loading_oid loading_get = app._loading.get
for oid in oid_list: for oid in oid_list:
invalidate(oid, tid) invalidate(oid, tid)
if oid == loading: loading = loading_get(oid)
app._loading_invalidated.append(tid) if loading:
loading[1].append(tid)
db = app.getDB() db = app.getDB()
if db is not None: if db is not None:
db.invalidate(tid, oid_list) db.invalidate(tid, oid_list)
finally: # ZODB<5: Update before releasing the lock so that app.load
app._cache_lock_release() # asks the last serial (with respect to already processed
# invalidations by Connection._setstate).
# ZODB≥5: Update after db.invalidate because the MVCC
# adapter starts at the greatest TID between
# IStorage.lastTransaction and processed invalidations.
app.last_tid = tid
def sendPartitionTable(self, conn, ptid, num_replicas, row_list): def sendPartitionTable(self, conn, ptid, num_replicas, row_list):
pt = self.app.pt = object.__new__(PartitionTable) pt = self.app.pt = object.__new__(PartitionTable)
......
...@@ -50,7 +50,7 @@ class Transaction(object): ...@@ -50,7 +50,7 @@ class Transaction(object):
self.conflict_dict = {} # {oid: serial} self.conflict_dict = {} # {oid: serial}
# resolved conflicts # resolved conflicts
self.resolved_dict = {} # {oid: serial} self.resolved_dict = {} # {oid: serial}
# involved storage nodes; connection is None is connection was lost # involved storage nodes; connection is None if connection was lost
self.conn_dict = {} # {node_id: connection} self.conn_dict = {} # {node_id: connection}
def __repr__(self): def __repr__(self):
......
...@@ -197,8 +197,7 @@ elif IF == 'trace-cache': ...@@ -197,8 +197,7 @@ elif IF == 'trace-cache':
@defer @defer
def profile(app): def profile(app):
app._cache_lock_acquire() with app._cache_lock:
try:
cache = app._cache cache = app._cache
if type(cache) is ClientCache: if type(cache) is ClientCache:
app._cache = CacheTracer(cache, '%s-%s.neo-cache-trace' % app._cache = CacheTracer(cache, '%s-%s.neo-cache-trace' %
...@@ -206,5 +205,3 @@ elif IF == 'trace-cache': ...@@ -206,5 +205,3 @@ elif IF == 'trace-cache':
app._cache.clear() app._cache.clear()
else: else:
app._cache = cache.close() app._cache = cache.close()
finally:
app._cache_lock_release()
...@@ -587,7 +587,7 @@ class Application(BaseApplication): ...@@ -587,7 +587,7 @@ class Application(BaseApplication):
node.send(Packets.StartOperation(self.backup_tid)) node.send(Packets.StartOperation(self.backup_tid))
uuid = node.getUUID() uuid = node.getUUID()
assert uuid not in self.storage_starting_set assert uuid not in self.storage_starting_set
if uuid not in self.storage_ready_dict: assert uuid not in self.storage_ready_dict
self.storage_starting_set.add(uuid) self.storage_starting_set.add(uuid)
def setStorageReady(self, uuid): def setStorageReady(self, uuid):
......
...@@ -65,6 +65,7 @@ There is no conflict of node id between the 2 clusters: ...@@ -65,6 +65,7 @@ There is no conflict of node id between the 2 clusters:
class BackupApplication(object): class BackupApplication(object):
pt = None pt = None
server = None # like in BaseApplication
uuid = None uuid = None
def __init__(self, app, name, master_addresses): def __init__(self, app, name, master_addresses):
......
...@@ -781,11 +781,19 @@ class MySQLDatabaseManager(DatabaseManager): ...@@ -781,11 +781,19 @@ class MySQLDatabaseManager(DatabaseManager):
if max_tid is not None: if max_tid is not None:
sql += " AND tid <= %d" % max_tid sql += " AND tid <= %d" % max_tid
q = self.query q = self.query
if q("SELECT 1 FROM trans%s LIMIT 1" % sql):
q("DELETE FROM trans" + sql) q("DELETE FROM trans" + sql)
else:
logging.info("Nothing to truncate in trans for partition %s",
partition)
sql = " FROM obj" + sql sql = " FROM obj" + sql
data_id_list = [x for x, in q( data_id_list = [x for x, in q(
"SELECT DISTINCT data_id%s AND data_id IS NOT NULL" % sql)] "SELECT DISTINCT data_id%s AND data_id IS NOT NULL" % sql)]
if q("SELECT 1%s LIMIT 1" % sql):
q("DELETE" + sql) q("DELETE" + sql)
else:
logging.info("Nothing to truncate in obj for partition %s",
partition)
self._pruneData(data_id_list) self._pruneData(data_id_list)
def getTransaction(self, tid, all = False): def getTransaction(self, tid, all = False):
......
...@@ -34,7 +34,7 @@ class ClientOperationHandler(BaseHandler): ...@@ -34,7 +34,7 @@ class ClientOperationHandler(BaseHandler):
app = self.app app = self.app
if app.operational: if app.operational:
# Even if in most cases, abortFor is called from both this method # Even if in most cases, abortFor is called from both this method
# and BaseMasterHandler.notifyPartitionChanges (especially since # and BaseMasterHandler.notifyNodeInformation (especially since
# storage nodes disconnects unknown clients on their own), these 2 # storage nodes disconnects unknown clients on their own), these 2
# handlers also cover distinct scenarios, so neither of them is # handlers also cover distinct scenarios, so neither of them is
# redundant: # redundant:
......
...@@ -139,6 +139,7 @@ class TransactionManager(EventQueue): ...@@ -139,6 +139,7 @@ class TransactionManager(EventQueue):
def replicating(self, offset_list): def replicating(self, offset_list):
self._replicating.update(offset_list) self._replicating.update(offset_list)
if __debug__:
isdisjoint = set(offset_list).isdisjoint isdisjoint = set(offset_list).isdisjoint
assert isdisjoint(self._replicated), (offset_list, self._replicated) assert isdisjoint(self._replicated), (offset_list, self._replicated)
assert isdisjoint(map(self.getPartition, self._store_lock_dict)), ( assert isdisjoint(map(self.getPartition, self._store_lock_dict)), (
......
...@@ -1084,8 +1084,7 @@ class NEOThreadedTest(NeoTestBase): ...@@ -1084,8 +1084,7 @@ class NEOThreadedTest(NeoTestBase):
def run(self): def run(self):
try: try:
apply(*self.__target) self.__result = apply(*self.__target)
self.__exc_info = None
except: except:
self.__exc_info = sys.exc_info() self.__exc_info = sys.exc_info()
if self.__exc_info[0] is NEOThreadedTest.failureException: if self.__exc_info[0] is NEOThreadedTest.failureException:
...@@ -1093,7 +1092,10 @@ class NEOThreadedTest(NeoTestBase): ...@@ -1093,7 +1092,10 @@ class NEOThreadedTest(NeoTestBase):
def join(self, timeout=None): def join(self, timeout=None):
threading.Thread.join(self, timeout) threading.Thread.join(self, timeout)
if not self.is_alive() and self.__exc_info: if not self.is_alive():
try:
return self.__result
except AttributeError:
etype, value, tb = self.__exc_info etype, value, tb = self.__exc_info
del self.__exc_info del self.__exc_info
raise etype, value, tb raise etype, value, tb
......
This diff is collapsed.
...@@ -349,6 +349,22 @@ class ReplicationTests(NEOThreadedTest): ...@@ -349,6 +349,22 @@ class ReplicationTests(NEOThreadedTest):
self.tic() self.tic()
self.assertTrue(backup.master.is_alive()) self.assertTrue(backup.master.is_alive())
@with_cluster(master_count=2)
def testBackupFromUpstreamWithSecondaryMaster(self, upstream):
"""
Check that the backup master reacts correctly when connecting first
to a secondary master of the upstream cluster.
"""
with NEOCluster(upstream=upstream) as backup:
primary = upstream.primary_master
m, = (m for m in upstream.master_list if m is not primary)
backup.master.resetNode(upstream_masters=[m.server])
backup.start()
backup.neoctl.setClusterState(ClusterStates.STARTING_BACKUP)
self.tic()
self.assertEqual(backup.neoctl.getClusterState(),
ClusterStates.BACKINGUP)
@backup_test() @backup_test()
def testCreationUndone(self, backup): def testCreationUndone(self, backup):
""" """
......
...@@ -39,6 +39,14 @@ class BasicTests(ZODBTestCase, StorageTestBase, BasicStorage): ...@@ -39,6 +39,14 @@ class BasicTests(ZODBTestCase, StorageTestBase, BasicStorage):
with Patch(threaded, TIC_LOOP=TIC_LOOP()): with Patch(threaded, TIC_LOOP=TIC_LOOP()):
super(BasicTests, self).check_checkCurrentSerialInTransaction() super(BasicTests, self).check_checkCurrentSerialInTransaction()
# The test expects that both load & lastTransaction would be blocked
# as long as the tpc_finish callback has not finished, taking more
# than .1 second. ZODB 5.6.0 clarified that lastTransaction() can
# return immediately with the previous last TID rather than blocking
# until it is allowed to return the new last TID.
check_tid_ordering_w_commit = unittest.skip("ZODB PR #316")(
BasicStorage.check_tid_ordering_w_commit)
if __name__ == "__main__": if __name__ == "__main__":
suite = unittest.makeSuite(BasicTests, 'check') suite = unittest.makeSuite(BasicTests, 'check')
unittest.main(defaultTest='suite') unittest.main(defaultTest='suite')
......
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