Commit 50d25d00 authored by Julien Muchembled's avatar Julien Muchembled

Drop 'background' mode completely in threaded tests

It was still used to stop a cluster.
parent 4253d24f
...@@ -44,6 +44,27 @@ BIND = IP_VERSION_FORMAT_DICT[ADDRESS_TYPE], 0 ...@@ -44,6 +44,27 @@ BIND = IP_VERSION_FORMAT_DICT[ADDRESS_TYPE], 0
LOCAL_IP = socket.inet_pton(ADDRESS_TYPE, IP_VERSION_FORMAT_DICT[ADDRESS_TYPE]) LOCAL_IP = socket.inet_pton(ADDRESS_TYPE, IP_VERSION_FORMAT_DICT[ADDRESS_TYPE])
class FairLock(deque):
"""Same as a threading.Lock except that waiting threads are queued, so that
the first one waiting for the lock is the first to get it. This is useful
when several concurrent threads fight for the same resource in loop:
the owner could give too little time for other to get a chance to acquire,
blocking them for a long time with bad luck.
"""
def __enter__(self, _allocate_lock=threading.Lock):
me = _allocate_lock()
me.acquire()
self.append(me)
other = self[0]
while me is not other:
with other:
other = self[0]
def __exit__(self, t, v, tb):
self.popleft().release()
class Serialized(object): class Serialized(object):
""" """
"Threaded" tests run all nodes in the same process as the test itself, "Threaded" tests run all nodes in the same process as the test itself,
...@@ -62,54 +83,20 @@ class Serialized(object): ...@@ -62,54 +83,20 @@ class Serialized(object):
we actually use Semaphores instead of Locks. we actually use Semaphores instead of Locks.
The epoll object of each node is hooked so that thread switching happens The epoll object of each node is hooked so that thread switching happens
before polling for network activity. An extra thread act as a scheduler before polling for network activity. An extra epoll object is used to
and uses an epoll object to detect which node has a readable epoll object. detect which node has a readable epoll object.
""" """
check_timeout = False check_timeout = False
class _Trigger(object):
_last = 0
def __init__(self):
self._lock = l = threading.Lock()
self._fd, w = os.pipe()
os.close(w)
l.acquire()
def __del__(self):
os.close(self._fd)
@classmethod @classmethod
def init(cls): def init(cls):
cls._background = 0
cls._busy = set() cls._busy = set()
cls._busy_cond = threading.Condition(threading.Lock()) cls._busy_cond = threading.Condition(threading.Lock())
cls._epoll = select.epoll() cls._epoll = select.epoll()
cls._pdb = None cls._pdb = None
cls._sched_lock = threading.Semaphore(0) cls._sched_lock = threading.Semaphore(0)
cls._step = -1 cls._tic_lock = FairLock()
cls._trigger = t = cls._Trigger() cls._fd_dict = {}
cls._fd_dict = {t._fd: t}
cls._thread = t = threading.Thread(target=cls._run, name=cls.__name__)
t.daemon = True
t.start()
@classmethod
def background(cls, background):
prev = cls._background
if prev != background:
if background:
cls._background = background
cls._sched_lock.release()
else:
fd = cls._trigger._fd
cls._epoll.register(fd)
cls._trigger._lock.acquire()
cls._epoll.unregister(fd)
cls.idle(None)
cls._background = background
return prev
@classmethod @classmethod
def idle(cls, owner): def idle(cls, owner):
...@@ -119,49 +106,17 @@ class Serialized(object): ...@@ -119,49 +106,17 @@ class Serialized(object):
@classmethod @classmethod
def stop(cls): def stop(cls):
assert cls._background assert not cls._fd_dict, cls._fd_dict
fd = cls._trigger._fd del(cls._busy, cls._busy_cond, cls._epoll, cls._fd_dict,
cls._epoll.register(fd) cls._pdb, cls._sched_lock, cls._tic_lock)
cls._trigger._lock.acquire()
del cls._trigger, cls._fd_dict[fd]
assert not cls._fd_dict
cls._sched_lock.release()
cls._thread.join()
del(cls._background, cls._busy, cls._busy_cond, cls._epoll,
cls._fd_dict, cls._pdb, cls._sched_lock, cls._step, cls._thread)
@classmethod @classmethod
def _run(cls): def _sort_key(cls, fd_event):
sched_lock = cls._sched_lock return -cls._fd_dict[fd_event[0]]._last
fd_dict = cls._fd_dict
sort_key = lambda fd_event: -fd_dict[fd_event[0]]._last
while 1:
sched_lock.acquire()
event_list = cls._step and cls._epoll.poll(0)
cls._step -= 1
if not event_list:
cls.idle(None)
if not cls._background:
continue
if not fd_dict:
break
event_list = cls._epoll.poll(-1)
cls._busy.add(None)
event_list.sort(key=sort_key)
next_lock = sched_lock
for fd, event in event_list:
self = fd_dict[fd]
self._release_next = next_lock.release
next_lock = self._lock
del self
next_lock.release()
@classmethod @classmethod
@contextmanager @contextmanager
def pdb(cls): def pdb(cls):
if cls._background:
yield
return
try: try:
cls._pdb = sys._getframe(2).f_trace.im_self cls._pdb = sys._getframe(2).f_trace.im_self
cls._pdb.set_continue() cls._pdb.set_continue()
...@@ -179,29 +134,35 @@ class Serialized(object): ...@@ -179,29 +134,35 @@ class Serialized(object):
def tic(cls, step=-1, check_timeout=()): def tic(cls, step=-1, check_timeout=()):
# If you're in a pdb here, 'n' switches to another thread # If you're in a pdb here, 'n' switches to another thread
# (the following lines are not supposed to be debugged into) # (the following lines are not supposed to be debugged into)
with cls.pdb(): # does nothing if background(1) was called with cls._tic_lock, cls.pdb():
f = sys._getframe(1) f = sys._getframe(1)
try: try:
logging.info('tic (%s:%u) ...', logging.info('tic (%s:%u) ...',
f.f_code.co_filename, f.f_lineno) f.f_code.co_filename, f.f_lineno)
finally: finally:
del f del f
if cls._background: if cls._busy:
assert step == -1 and not check_timeout
else:
with cls._busy_cond: with cls._busy_cond:
while cls._busy: while cls._busy:
cls._busy_cond.wait() cls._busy_cond.wait()
cls._busy.add(None)
cls._step = step
for app in check_timeout: for app in check_timeout:
app.em.epoll.check_timeout = True app.em.epoll.check_timeout = True
app.em.wakeup() app.em.wakeup()
del app del app
cls._sched_lock.release() while step:
with cls._busy_cond: event_list = cls._epoll.poll(0)
while cls._busy: if not event_list:
cls._busy_cond.wait() break
step -= 1
event_list.sort(key=cls._sort_key)
next_lock = cls._sched_lock
for fd, event in event_list:
self = cls._fd_dict[fd]
self._release_next = next_lock.release
next_lock = self._lock
del self
next_lock.release()
cls._sched_lock.acquire()
def __init__(self, app, busy=True): def __init__(self, app, busy=True):
self._epoll = app.em.epoll self._epoll = app.em.epoll
...@@ -251,7 +212,7 @@ class TestSerialized(Serialized): ...@@ -251,7 +212,7 @@ class TestSerialized(Serialized):
Serialized.__init__(busy=False, *args) Serialized.__init__(busy=False, *args)
def poll(self, timeout): def poll(self, timeout):
if timeout and not self._background: if timeout:
while 1: while 1:
r = self._epoll.poll(0) r = self._epoll.poll(0)
if r: if r:
...@@ -584,12 +545,11 @@ class NEOCluster(object): ...@@ -584,12 +545,11 @@ class NEOCluster(object):
Serialized.init() Serialized.init()
@staticmethod @staticmethod
def _unpatch(background): def _unpatch():
cls = NEOCluster cls = NEOCluster
assert cls._patch_count > 0 assert cls._patch_count > 0
cls._patch_count -= 1 cls._patch_count -= 1
if cls._patch_count: if cls._patch_count:
Serialized.background(background)
return return
BaseConnection.getTimeout = cls.BaseConnection_getTimeout BaseConnection.getTimeout = cls.BaseConnection_getTimeout
SimpleQueue.__init__ = cls.SimpleQueue__init__ SimpleQueue.__init__ = cls.SimpleQueue__init__
...@@ -735,21 +695,23 @@ class NEOCluster(object): ...@@ -735,21 +695,23 @@ class NEOCluster(object):
self._db = db = ZODB.DB(storage=self.getZODBStorage()) self._db = db = ZODB.DB(storage=self.getZODBStorage())
return db return db
def join(self, thread_list, timeout=5):
timeout += time.time()
while thread_list:
assert time.time() < timeout
Serialized.tic()
thread_list = [t for t in thread_list if t.is_alive()]
def stop(self): def stop(self):
logging.debug("stopping %s", self) logging.debug("stopping %s", self)
background = Serialized.background(True)
self.__dict__.pop('_db', self.client).close() self.__dict__.pop('_db', self.client).close()
node_list = self.admin_list + self.storage_list + self.master_list node_list = self.admin_list + self.storage_list + self.master_list
for node in node_list: for node in node_list:
node.em.wakeup(True) node.em.wakeup(True)
for node in node_list: node_list.append(self.client.poll_thread)
if node._Thread__started.is_set(): self.join(node_list)
node.join()
client = self.client.poll_thread
if client.is_alive():
client.join()
logging.debug("stopped %s", self) logging.debug("stopped %s", self)
self._unpatch(background) self._unpatch()
def getNodeState(self, node): def getNodeState(self, node):
uuid = node.uuid uuid = node.uuid
......
...@@ -533,16 +533,10 @@ class Test(NEOThreadedTest): ...@@ -533,16 +533,10 @@ class Test(NEOThreadedTest):
t.commit() t.commit()
# tell admin to shutdown the cluster # tell admin to shutdown the cluster
cluster.neoctl.setClusterState(ClusterStates.STOPPING) cluster.neoctl.setClusterState(ClusterStates.STOPPING)
self.tic()
# all nodes except clients should exit # all nodes except clients should exit
for master in cluster.master_list: cluster.join(cluster.master_list
master.join(5) + cluster.storage_list
self.assertFalse(master.is_alive()) + cluster.admin_list)
for storage in cluster.storage_list:
storage.join(5)
self.assertFalse(storage.is_alive())
cluster.admin.join(5)
self.assertFalse(cluster.admin.is_alive())
finally: finally:
cluster.stop() cluster.stop()
cluster.reset() # reopen DB to check partition tables cluster.reset() # reopen DB to check partition tables
......
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