Francesco Romani has uploaded a new change for review.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
threadpool: add a thread pool with watchdog
Add a new thread pool implementation with built-in watchdog.
The watchdog is needed to accomodate the needs of the virt sampling code. The virt sampling needs to deal with possibly blocking I/O like calls which cannot be made not-blocking.
Change-Id: Ic06da1ba57868dc2c7db67a1868ad10087a1cff2 Signed-off-by: Francesco Romani fromani@redhat.com --- M lib/threadpool/Makefile.am A lib/threadpool/examples/demo.py A lib/threadpool/schedqueue.py A lib/threadpool/tests/test_schedqueue.py A lib/threadpool/tests/test_timetrackingthread.py A lib/threadpool/tests/test_worker.py A lib/threadpool/tests/test_workqueue.py A lib/threadpool/watchedpool.py A lib/threadpool/watchman.py A lib/threadpool/worker.py A lib/threadpool/workqueue.py 11 files changed, 1,003 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/91/29191/1
diff --git a/lib/threadpool/Makefile.am b/lib/threadpool/Makefile.am index 3a71a5d..8b7098e 100644 --- a/lib/threadpool/Makefile.am +++ b/lib/threadpool/Makefile.am @@ -22,4 +22,9 @@
dist_threadpool_PYTHON = \ __init__.py \ + schedqueue.py \ + watchedpool.py \ + watchman.py \ + worker.py \ + workqueue.py \ $(NULL) diff --git a/lib/threadpool/examples/demo.py b/lib/threadpool/examples/demo.py new file mode 100644 index 0000000..cf80930 --- /dev/null +++ b/lib/threadpool/examples/demo.py @@ -0,0 +1,62 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import logging +import random +import sys +import time + +from watchedpool import WatchedThreadPool + + +def _test_fun(*args): + """ + just block the caller for a random time span. + """ + tmo = random.randint(1, 15) + logging.info('started, waiting: %.3fs', tmo) + time.sleep(tmo) + logging.info('done') + + +def _main(args): + """ + main test driver + """ + nworkers = int(args[0]) if len(args) >= 1 else 1 + pool = WatchedThreadPool(nworkers) + pool.start() + for tag in range(nworkers): + pool.submit_periodic(4, _test_fun) + + try: + while True: + time.sleep(1.0) + except KeyboardInterrupt: + logging.debug('exiting!') + finally: + pool.stop() + + +if __name__ == '__main__': + logging.basicConfig( + level=logging.DEBUG, + format='%(asctime)s %(threadName)-10s %(levelname)-8s %(message)s') + _main(sys.argv[1:]) diff --git a/lib/threadpool/schedqueue.py b/lib/threadpool/schedqueue.py new file mode 100644 index 0000000..c2bfbae --- /dev/null +++ b/lib/threadpool/schedqueue.py @@ -0,0 +1,123 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +""" +Time-aware priority queue, built like queue.Queue +but tailored for WatchedThreadPool needs. +""" + +import heapq +import threading +import time + + +class QueueError(Exception): + """ + Queue exception root. + """ + + +class Empty(QueueError): + """ + Attempt to get an item from an empty queue. + """ + + +class Full(QueueError): + """ + Attempt to put an item into a full queue. + """ + + +class NotYet(QueueError): + """ + Attempt to get an item which is not due at the + present (reported) time. + """ + + +class SchedQueue(object): + """ + Time-aware, not blocking (client must take care of + waiting/timing) priority queue. + """ + def __init__(self, maxsize=0, timefunc=time.time): + self._timefunc = timefunc + self._maxsize = maxsize + self._queue = self._init(maxsize) + self._lock = threading.Lock() + + def __len__(self): + with self._lock: + return self._size() + + def put(self, item, delay=0): + """ + Put an item into the queue, to be made ready 'delay' seconds + in the future. + """ + with self._lock: + if self._maxsize > 0 and self._size() == self._maxsize: + raise Full + self._put((self._timefunc() + delay, item)) + + def get(self, timenow): + """ + Ask for an item to be consumed. + """ + with self._lock: + if not self._size(): + raise Empty + timeval, _ = self._peek() + if timeval > timenow: + raise NotYet + _, item = self._get() + return item + + def _init(self, maxsize): + """ + builds the internal queue. + """ + self._maxsize = maxsize # to make pylint happy + return [] + + def _size(self): + """ + size of the internal queue. + """ + return len(self._queue) + + def _put(self, item): + """ + put a new element (at the end of) in the queue. + """ + heapq.heappush(self._queue, item) + + def _get(self): + """ + gets (and removes) the next element in queue. + """ + return heapq.heappop(self._queue) + + def _peek(self): + """ + read-only view of the next element in queue. + """ + return self._queue[0] diff --git a/lib/threadpool/tests/test_schedqueue.py b/lib/threadpool/tests/test_schedqueue.py new file mode 100644 index 0000000..3595589 --- /dev/null +++ b/lib/threadpool/tests/test_schedqueue.py @@ -0,0 +1,56 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import unittest + +import schedqueue + + +class SchedQueueTests(unittest.TestCase): + def _faketime(self): + ret = self._ts + self._ts += 1 + return ret + + def setUp(self): + self._ts = 0 + self._sq = schedqueue.SchedQueue(timefunc=self._faketime) + + def test_empty_on_create(self): + self.assertEqual(len(self._sq), 0) + + def test_put_on_full(self): + self._sq = schedqueue.SchedQueue(maxsize=1, timefunc=self._faketime) + self._sq.put('foo') + self.assertRaises(schedqueue.Full, self._sq.put, 'bar') + + def test_get_on_empty(self): + self.assertRaises(schedqueue.Empty, self._sq.get, 1) + + def test_get_too_early(self): + item = 'foo' + self._sq.put(item, delay=5) + self.assertRaises(schedqueue.NotYet, self._sq.get, 0) + + def test_put_get(self): + item = 'foo' + self._sq.put(item) + res = self._sq.get(timenow=1) + self.assertEqual(res, item) diff --git a/lib/threadpool/tests/test_timetrackingthread.py b/lib/threadpool/tests/test_timetrackingthread.py new file mode 100644 index 0000000..14a25ef --- /dev/null +++ b/lib/threadpool/tests/test_timetrackingthread.py @@ -0,0 +1,64 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import unittest + +import worker + + +class TimeTrackingThreadTests(unittest.TestCase): + def _fakesleep(self, delay): + self._ts += max(0, delay) # avoid negative delays + + def _faketime(self): + ret = self._ts + self._ts += 1 + return ret + + def setUp(self): + self._ts = 0 + self._ttt = worker.TimeTrackingThread(self._faketime) + + def assertIdle(self): + self.assertFalse(self._ttt.busy) + self.assertEqual(self._ttt.task, None) + + def assertBusyOn(self, task): + self.assertTrue(self._ttt.busy) + self.assertEqual(self._ttt.task, task) + + def test_created_free(self): + self.assertIdle() + + def test_carry_task(self): + self.assertIdle() + + task = 'foobar' + with self._ttt.track_time(task): + self.assertBusyOn(task) + + self.assertIdle() + + def test_elapsed(self): + task = 'barbaz' + delay = 5 + with self._ttt.track_time(task): + self._fakesleep(delay) + self.assertTrue(self._ttt.elapsed >= delay) diff --git a/lib/threadpool/tests/test_worker.py b/lib/threadpool/tests/test_worker.py new file mode 100644 index 0000000..7396673 --- /dev/null +++ b/lib/threadpool/tests/test_worker.py @@ -0,0 +1,60 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import contextlib +import unittest + +import schedqueue +import worker + + +class FakeWorkQueue(object): + def __init__(self, items): + self._items = items + + @contextlib.contextmanager + def fetch(self): + if not self._items: + raise schedqueue.Empty + item = self._items.pop(0) + if issubclass(item, Exception): + raise item + else: + yield item + + +class WorkerThreadTests(unittest.TestCase): + def test__pull_empty(self): + wq = FakeWorkQueue([]) + wt = worker.Worker(wq) + with wt._pull() as item: + self.assertEqual(item, (None, None, None, None)) + + def test__pull_empty_exc(self): + wq = FakeWorkQueue([schedqueue.Empty]) + wt = worker.Worker(wq) + with wt._pull() as item: + self.assertEqual(item, (None, None, None, None)) + + def test__pull_notyet(self): + wq = FakeWorkQueue([schedqueue.NotYet]) + wt = worker.Worker(wq) + with wt._pull() as item: + self.assertEqual(item, (None, None, None, None)) diff --git a/lib/threadpool/tests/test_workqueue.py b/lib/threadpool/tests/test_workqueue.py new file mode 100644 index 0000000..68e982d --- /dev/null +++ b/lib/threadpool/tests/test_workqueue.py @@ -0,0 +1,80 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import unittest + +import schedqueue +import watchedpool + + +def _do_nothing(*args, **kwargs): + pass + + +class WorkQueueTests(unittest.TestCase): + def _fakesleep(self, delay): + pass + + def _faketime(self): + ret = self._ts + self._ts += 1 + return ret + + def assertDummyWork(self, workitem, in_tag=None): + out_tag, work, args, kwargs = workitem + if in_tag is not None: + self.assertEqual(in_tag, out_tag) + self.assertEqual(work, _do_nothing) + self.assertEqual(len(args), 0) + self.assertEqual(len(kwargs), 0) + + def setUp(self): + self._ts = 0 + self._wq = watchedpool.WorkQueue( + timefunc=self._faketime, + delayfunc=self._fakesleep) + + def test_created_empty(self): + self.assertEqual(len(self._wq), 0) + + def test_get_on_empty(self): + self.assertRaises(schedqueue.Empty, self._wq.get) + + def test_post_on_empty(self): + self._wq.post(0, _do_nothing) + self.assertEqual(len(self._wq), 1) + + def test_put_get(self): + in_tag = 'foo' + self._wq.put(in_tag, 0, _do_nothing) + self.assertDummyWork(self._wq.get()) + + def test_remove(self): + in_tag = 'foo' + self._wq.put(in_tag, 0, _do_nothing) + self._wq.remove(in_tag) + self.assertRaises(schedqueue.Empty, self._wq.get) + + def test_fetch(self): + in_tag = 'foo' + self._wq.put(in_tag, 1, _do_nothing) + with self._wq.fetch() as work_item: + self.assertDummyWork(work_item) + self.assertEqual(len(self._wq), 1) diff --git a/lib/threadpool/watchedpool.py b/lib/threadpool/watchedpool.py new file mode 100644 index 0000000..6a8938f --- /dev/null +++ b/lib/threadpool/watchedpool.py @@ -0,0 +1,128 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +""" +PoC of a watched thread pool implementation. +This thread pool adds a watcher/watchman thread which detaches `rogue' +threads from the main pool, and ask them to die as soon as possible. +The purpose is to avoid worker threads getting stuck too much on +busy, possible blocking operations (e.g. risky I/O). + +In the worst case scenario, the pool size degenerates in a +thread-per-operation. In the common and expected case, +the pool size is fixed and rogue threads are just spikes +in the thread number. + +TODO: +- allow the client to resize dynamically the pool [?] +- let this code be easily mergeable with vdsm/storage/threadPool.py +- demonstrate that this code does not leak threads. +""" + +import logging +import threading + +from watchman import Watchman +from worker import Worker +from workqueue import WorkQueue + + +class WatchedThreadPool(object): + """ + Straightforward thread pool, with an added Watchman. + """ + def __init__(self, nworkers=1, work_interval=1): + self._nworkers = nworkers + self._work_interval = work_interval + self._watch_interval = 2 * self._work_interval + self._busy_timeout = 3 * self._watch_interval + self._work_queue = None + self._workers = None + self._watcher = None + self._running = False + self._lock = threading.Lock() + + def start(self): + """ + starts the pool. Thread safe. + """ + with self._lock: + if not self._running: + self._work_queue = WorkQueue(self._work_interval) + self._workers = [self._make_worker() + for _ in range(self._nworkers)] + self._watcher = Watchman(self._workers, + self._make_worker, + self._watch_interval, + self._busy_timeout) + self._watcher.start() + logging.info('pool started') + self._running = True + + def stop(self, wait=True): + """ + terminate all pooled threads. Thread safe. + """ + with self._lock: + if self._running: + self._running = False + + self._watcher.stop() + if wait: + for worker in self._workers: + worker.join() + del worker + self._watcher.reap([]) # last chance + self._watcher.join() + del self._watcher + + def submit_periodic(self, interval, work, *args, **kwargs): + """ + submit a new work unit to the pool. + returns the work unit tag. + The tag is guaranteed to be unique per-pool. + """ + if self._running: + tag = self._work_queue.post(interval, work, args, kwargs) + logging.debug('new work: (%s, %s, %s) -> %s', + str(work), str(args), str(kwargs), tag) + return tag # FIXME + else: + return None + + def remove(self, tag): + """ + remove a work unit from the pool. + """ + return self._work_queue.remove(tag) + + def notify_timeout(self, tag, on_timeout): + """ + register a timeout callback for the given work unit. + """ + self._watcher.notify_timeout(tag, on_timeout) + + def _make_worker(self): + """ + creates a new idle (not started) worker. + """ + worker = Worker(self._work_queue) + logging.info('worker %s added to the pool', worker.name) + return worker diff --git a/lib/threadpool/watchman.py b/lib/threadpool/watchman.py new file mode 100644 index 0000000..53bfc11 --- /dev/null +++ b/lib/threadpool/watchman.py @@ -0,0 +1,209 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +""" +PoF of a watched thread pool implementation. +This thread pool adds a watcher/watchman thread which detaches `rogue' +threads from the main pool, and ask them to die as soon as possible. +The purpose is to avoid worker threads getting stuck too much on +busy, possible blocking operations (e.g. risky I/O). + +In the worst case scenario, the pool size degenerates in a +thread-per-operation. In the common and expected case, +the pool size is fixed and rogue threads are just spikes +in the thread number. + +TODO: +- allow the client to resize dynamically the pool [?] +- let this code be easily mergeable with vdsm/storage/threadPool.py +- demonstrate that this code does not leak threads. +""" + +import logging +import threading +import time + + +class Watchman(threading.Thread): + """ + Watches the worker threads, detects the rogues, if any, + and possibly replace them to ensure the pool is operative + and not blocked. + """ + def __init__(self, workers, make_worker, + watch_interval=5, busy_timeout=10, + rogues_cap=None, delayfunc=time.sleep): + super(Watchman, self).__init__() + self.daemon = True + self._workers = workers + self._rogues = [] + self._rogues_cap = rogues_cap + self._watch_interval = watch_interval + self._make_worker = make_worker + self._busy_timeout = busy_timeout + self._timeout_cb = {} + self._stop = threading.Event() + self._running = True + self._delayfunc = delayfunc + + def suspend(self): + """ + suspend the watch, without stopping. + """ + self._running = False + + def resume(self): + """ + resume the watch. + """ + self._running = True + + def start(self): + """ + start the watched pool and the watcher itself. + """ + for worker in self._workers: + worker.start() + super(Watchman, self).start() + + def stop(self): + """ + stop the watched pool, and the watcher itself. + """ + self.suspend() + for worker in self._workers: + worker.stop() + self._stop.set() + + def run(self): + logging.debug( + 'watcher starting (with %i workers)', len(self._workers)) + + while not self._stop.is_set(): + self._delayfunc(self._watch_interval) + if self._running: + workers, rogues, newbies = self.verify() + self.enforce(workers, rogues, newbies) + + logging.debug('watcher done') + + def notify_timeout(self, tag, on_timeout): + """ + register a timeout callback for the given work unit. + """ + self._timeout_cb[tag] = on_timeout + + def stuck(self, worker): + """ + is a given worker stuck? + """ + stuck, task = False, None + if worker.busy: + # use locals to freeze the state + task = worker.task + elapsed = worker.elapsed + stuck = elapsed > self._busy_timeout + logging.warning( + '%s busy on %s for the last %02i seconds (max=%02i) -> %s', + worker.name, task, elapsed, self._busy_timeout, + 'STUCK' if stuck else 'OK') + return stuck, task + + def verify(self): + """ + check the health of the worker pool, and classify + it into + - workers: well behaving worker threads + - rogues: workers got stuck, to be evicted from the pool + - newbies: new workers ready to replace rogues + """ + try: + workers, rogues, newbies = [], [], [] + logging.debug('verification loop started') + for worker in self._workers: + stuck, task = self.stuck(worker) + if not stuck: + workers.append(worker) + elif not self._room_for_rogues(rogues): + logging.warning( + 'worker %s gone rogue, but cap reached', worker.name) + elif not self._is_detachable(task): + logging.warning( + 'worker %s gone rogue, but task %s not detachable', + worker.name, task) + else: + logging.info( + 'worker %s gone rogue, replacing', worker.name) + worker.stop() + rogues.append(worker) # until it dies + newbies.append(self._make_worker()) + logging.debug('verification loop done: %i rogues', len(rogues)) + return workers, rogues, newbies + except Exception: + logging.exception('verify failed', exc_info=True) + + def enforce(self, workers, rogues, newbies): + """ + enforce the pool size by evicting rogues into a limbo, + and by replenishing the worker pool with the provided + newbies. + """ + try: + for newbie in newbies: + newbie.start() + workers.append(newbie) + self._workers = workers + self.reap(rogues) + except Exception: + logging.exception('enforce failed', exc_info=True) + + def reap(self, rogues): + """ + collect the terminated rogues. + """ + rogues.extend(self._rogues) + new_rogues = [] + for rogue in rogues: + if rogue.is_alive(): + new_rogues.append(rogue) + else: + logging.info('rogue thread %s collected', rogue.name) + rogue.join() + del rogue + self._rogues = new_rogues + + def _room_for_rogues(self, rogues): + return (self._rogues_cap is None + or len(rogues) < self._rogues_cap) + + def _is_detachable(self, task): + """ + if a task is detachable, its worker thread will be detached + by the pool. Otherwise the thread will be kept in the pool + """ + on_timeout = self._timeout_cb.get(task, _on_timeout_dummy) + return on_timeout(task) + + +def _on_timeout_dummy(*args): + """ + Dummy timeout callback. Marks every work unit as detachable. + """ + return True diff --git a/lib/threadpool/worker.py b/lib/threadpool/worker.py new file mode 100644 index 0000000..fe63d34 --- /dev/null +++ b/lib/threadpool/worker.py @@ -0,0 +1,129 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import contextlib +import logging +import threading +import time + +import schedqueue + + +class TimeTrackingThread(threading.Thread): + """ + Thin mixin to help a worker thread to track the time + consumed by a work item. + """ + def __init__(self, timefunc=time.time): + super(TimeTrackingThread, self).__init__() + self._work_started_at = None + self._working_on = None + self._timefunc = timefunc + + @property + def busy(self): + """ + is the thread doing some work or is it idle? + """ + return self._work_started_at is not None + + @property + def elapsed(self): + """ + how much time since the thread started the current work unit? + """ + return self._timefunc() - self._work_started_at + + @property + def task(self): + """ + what is this worker doing? + """ + return self._working_on + + @contextlib.contextmanager + def track_time(self, tag): + """ + helper context manager to track the time consumed in + the current work unit + """ + self._working_on = tag + self._work_started_at = self._timefunc() + yield + self._work_started_at = None + self._working_on = None + + +class Worker(TimeTrackingThread): + """ + A regular worker thread. This pool has no additional + requirements saved the ability of a worker thread + to die ASAP (see the stop() method) + """ + def __init__(self, work_queue): + super(Worker, self).__init__() + self.daemon = True + self._work_queue = work_queue + self._stop = threading.Event() + + def run(self): + logging.info('worker %s starting', self.name) + + while not self._stop.is_set(): + self.process() + + logging.info('worker %s done', self.name) + + def process(self): + """ + execute a single try of work processing. + """ + with self._pull() as (tag, work, args, kwargs): + if work is not None: + self._do_work(tag, work, args, kwargs) + + def stop(self): + """ + gracefully terminate ASAP, and avoid to begin more work. + """ + self._stop.set() + + @contextlib.contextmanager + def _pull(self): + """ + pulls the next work item + """ + try: + with self._work_queue.fetch() as item: + yield item + except schedqueue.NotYet: + yield None, None, None, None + except schedqueue.Empty: + yield None, None, None, None + + def _do_work(self, tag, work, args, kwargs): + """ + process a single work unit. + """ + with self.track_time(tag): + try: + work(*args, **kwargs) + except Exception: + logging.exception('work failed', exc_info=True) diff --git a/lib/threadpool/workqueue.py b/lib/threadpool/workqueue.py new file mode 100644 index 0000000..73949b8 --- /dev/null +++ b/lib/threadpool/workqueue.py @@ -0,0 +1,87 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import contextlib +import time +import uuid + +import schedqueue + + +class WorkQueue(object): + """ + Tiny wrapper around a plain Queue.Queue, to encapsulate + the 'periodic work' behaviour. + """ + def __init__(self, wait_timeout=1, + timefunc=time.time, delayfunc=time.sleep): + self._queue = schedqueue.SchedQueue(timefunc=timefunc) + self._work = {} + self._wait_timeout = wait_timeout + self._timefunc = timefunc + self._delayfunc = delayfunc + + def put(self, tag, interval, work, *args, **kwargs): + """ + like Queue.put(). + """ + item = (work, args, kwargs, interval) + self._work[tag] = item + self._queue.put(tag, delay=interval) + + def get(self): + """ + like Queue.get(). + """ + tag, work, args, kwargs, _ = self._get() + return (tag, work, args, kwargs) + + def remove(self, tag): + return self._work.pop(tag, None) + + def post(self, interval, work, *args, **kwargs): + """ + like put(), but with auto-tagging. + """ + return self.put(str(uuid.uuid4()), interval, work, *args, **kwargs) + + @contextlib.contextmanager + def fetch(self): + """ + like get(), but automatically reinject the work + to do in the queue if the task is periodic. + """ + tag, work, args, kwargs, interval = self._get() + yield (tag, work, args, kwargs) + if interval: + self.put(tag, interval, work, args, kwargs) + + def _get(self): + tag, work, args, kwargs = None, None, None, None + while work is None: + self._delayfunc(self._wait_timeout) + tag = self._queue.get(self._timefunc()) + item = self._work.get(tag, None) + if item is not None: + work, args, kwargs, interval = item + return (tag, work, args, kwargs, interval) + + def __len__(self): + return len(self._work)
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 1:
Build Successful
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9631/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10415/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit/10572/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_verify-error-codes_merged/5496/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_merged/3654/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 2:
Build Successful
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9632/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10417/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit/10574/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_verify-error-codes_merged/5498/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_merged/3656/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 3:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9785/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10570/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/35/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/68/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/47/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/10727/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 4: Code-Review-1 Verified-1
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9789/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10574/ : UNSTABLE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/38/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/71/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/50/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/10731/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 5: Code-Review-1 Verified-1
Build Unstable
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9791/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10576/ : UNSTABLE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/40/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/73/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/52/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/10733/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 6: Code-Review-1 Verified-1
Build Unstable
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9794/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10579/ : UNSTABLE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/43/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/76/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/55/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/10736/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 7:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9805/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10590/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/49/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/82/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/61/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/10747/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9835/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10620/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/57/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/90/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/69/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/10777/ : FAILURE
Nir Soffer has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(11 comments)
On first look in the worker module, this is over-engineered and too complex. From the description, we need a simple pool of threads with an extra supervisor thread to handle stuck tasks.
We need the simplest thread pool that is good only for what vdsm needs and nothing else.
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 25: Line 26: import schedqueue Line 27: Line 28: Line 29: class TimeTrackingThread(threading.Thread): Please don't add unneeded inheritance. If we need time tracking, add it to the worker thread class.
And specially do not inherit from Thread - instead keep a thread instance varibale. This allows us to implement the interface that we like to have, and we do not depend on Thread. Line 30: """ Line 31: Thin mixin to help a worker thread to track the time Line 32: consumed by a work item. Line 33: """
Line 37: self._working_on = None Line 38: self._timefunc = timefunc Line 39: Line 40: @property Line 41: def busy(self): Not related to time tracking Line 42: """ Line 43: is the thread doing some work or is it idle? Line 44: """ Line 45: return self._work_started_at is not None
Line 55: def task(self): Line 56: """ Line 57: what is this worker doing? Line 58: """ Line 59: return self._working_on naming it _task is less surprizing Line 60: Line 61: @contextlib.contextmanager Line 62: def track_time(self, tag): Line 63: """
Line 67: self._working_on = tag Line 68: self._work_started_at = self._timefunc() Line 69: yield Line 70: self._work_started_at = None Line 71: self._working_on = None Not needed. The _do_work of the worker can simply do:
start = self.clock.time() try: run task... finally: self.elapsed = self.clock.time() - start
If you want a generic time tracking utility, put it in utils.py:
class StopWatch:
def __init__(self): self.start = None self.elapsed = None
def __enter__(self): self.start = time.time() return self
def __exit__(self, *args): self.elapsed = time.time() - self.start
with StopWatch() as stopwatch: do stuff...
And use it:
logging.debug("did stuff in %.2f seconds", stopwatch.elapsed)
We don't need complex class hierarchy for this. Line 72: Line 73: Line 74: class Worker(TimeTrackingThread): Line 75: """
Line 79: """ Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Whats wrong with the builtin Queue? Line 84: self._stop = threading.Event() Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name)
Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event() This allows stopping immediately - but usually you want to stop when work in the queue is done. Also this adds another lock and condition for each worker, while we can have only single lock and condition for the entire thread pool if we use the method I describe bellow.
To stop a worker, put a a sentinel object into the queue:
_STOP = object()
In ThreadPool:
def stop_some_worker(self): self._queue.put(_STOP)
In Worker:
def _run(self): while self._running: self._process()
def _process(self): it = self._queue.get() if it is _STOP: self._running = False return process it...
If I want to stop N workers, I add N _STOP object into the queue. If I want to stop immediately, I can add the _STOP objects at the end of the queue. If I want to stop when current work is done, I can add them at the front of the queue.
We may need to cancel a stuck thread, but for this the stop event does not help. We need a way to unblock a stuck task in a way that will raise an exception that cause the thread to exit cleanly, or just fail the current task.
For example, if the current task is stuck on a socket, we can make it fail and let the thread continue to process the next request. Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88:
Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event() Line 85: Use @traceback Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88: Line 89: while not self._stop.is_set():
Line 95: """ Line 96: execute a single try of work processing. Line 97: """ Line 98: with self._pull() as item: Line 99: if item is not None: Why do you want to support None? Line 100: self._do_work(item) Line 101: Line 102: def stop(self): Line 103: """
Line 115: yield item Line 116: except schedqueue.NotYet: Line 117: yield None Line 118: except schedqueue.Empty: Line 119: yield None Unneeded complexity. Line 120: Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit.
Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit. Line 124: """ Line 125: tag, work, args, kwargs = item work is a function or callable - why not call use an obvious name? Line 126: with self.track_time(tag): Line 127: try: Line 128: work(*args, **kwargs) Line 129: except Exception:
Line 126: with self.track_time(tag): Line 127: try: Line 128: work(*args, **kwargs) Line 129: except Exception: Line 130: logging.exception('work failed', exc_info=True) Don't use the non-standard exec_info, exception logs a traceback anyway.
Francesco Romani has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(8 comments)
Nir, thanks for the review. First batch of answers, more to come.
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 25: Line 26: import schedqueue Line 27: Line 28: Line 29: class TimeTrackingThread(threading.Thread):
Please don't add unneeded inheritance. If we need time tracking, add it to
Good points, will change. Line 30: """ Line 31: Thin mixin to help a worker thread to track the time Line 32: consumed by a work item. Line 33: """
Line 37: self._working_on = None Line 38: self._timefunc = timefunc Line 39: Line 40: @property Line 41: def busy(self):
Not related to time tracking
will move away from here, probably into worker thread Line 42: """ Line 43: is the thread doing some work or is it idle? Line 44: """ Line 45: return self._work_started_at is not None
Line 55: def task(self): Line 56: """ Line 57: what is this worker doing? Line 58: """ Line 59: return self._working_on
naming it _task is less surprizing
Agreed. will change. Line 60: Line 61: @contextlib.contextmanager Line 62: def track_time(self, tag): Line 63: """
Line 67: self._working_on = tag Line 68: self._work_started_at = self._timefunc() Line 69: yield Line 70: self._work_started_at = None Line 71: self._working_on = None
Not needed. The _do_work of the worker can simply do:
Agreed on the general concept -and agreed that this can be simplified: will do.
I cannot directly adopt your suggestion because the worker thread doesn't need to know how much time a task took (nor the supervisor does), but instead the supervisor will check asynchronously how much time a task is *taking* (while still in progress) in order to implement to detect stuck task (= task being processed for a time exceeding a threshold).
However there is definitely room for simplification, and will do. Line 72: Line 73: Line 74: class Worker(TimeTrackingThread): Line 75: """
Line 79: """ Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue
Whats wrong with the builtin Queue?
The standard library provides a nice priorityqueue, which is almost what I need here, except for the fact it has not the notion of time, which I need to implement the periodic tasks for interval sampling (poll X every Y seconds).
On the other hand, there is the sched module: https://docs.python.org/2/library/sched.html which has the notion of time but it is not thread safe and has a blocking interface (see the delayfunc argument).
I need to blend these concepts together somehow, then the workqueue/schedqueue classes are born. Line 84: self._stop = threading.Event() Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name)
Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event() Line 85:
Use @traceback
Will do Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88: Line 89: while not self._stop.is_set():
Line 115: yield item Line 116: except schedqueue.NotYet: Line 117: yield None Line 118: except schedqueue.Empty: Line 119: yield None
Unneeded complexity.
I have to agree. All this dance is done to let the queue transparently reinject a periodic task into itself.
I'm starting to think to just simply do this in the worker thread and save quite some complexity. Line 120: Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit.
Line 126: with self.track_time(tag): Line 127: try: Line 128: work(*args, **kwargs) Line 129: except Exception: Line 130: logging.exception('work failed', exc_info=True)
Don't use the non-standard exec_info, exception logs a traceback anyway.
will do
Nir Soffer has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(3 comments)
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 67: self._working_on = tag Line 68: self._work_started_at = self._timefunc() Line 69: yield Line 70: self._work_started_at = None Line 71: self._working_on = None
Agreed on the general concept -and agreed that this can be simplified: will
I'm not sure that detecting stuck thread by tracking time will work. Different tasks have different timeouts, so stuck state must be a state of the task, not the worker thread or the supervisor.
We can use a supervisor thread that check task state, and try to cancel/remove thread when task report it is stuck. Task will have to track time if stuck state depends on a timeout.
Another solution may be that each task has a user waiting on the task (e.g. waiting on Event object). If the user of the task time out, it can cancel the task, which can raise Canceled exception, waking up the thread and dropping the task. If we go in this way, we don't need to change the size of the thread pool while its running, because thread cannot get stuck.
I suggest we don't design a generic thread pool but build a libvirt connection pool which is what we currently need. When this is done and if we have duplicate code we can remove duplication by creating a common thread pool. Line 72: Line 73: Line 74: class Worker(TimeTrackingThread): Line 75: """
Line 79: """ Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue
The standard library provides a nice priorityqueue, which is almost what I
Why do you need priorityqueue or scheduler? Line 84: self._stop = threading.Event() Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name)
Line 115: yield item Line 116: except schedqueue.NotYet: Line 117: yield None Line 118: except schedqueue.Empty: Line 119: yield None
I have to agree. All this dance is done to let the queue transparently rein
I don't understand why you need to inject event into the queue. I would implement the supervisor thread in another unrelated thread, using simple loop and Event.wait(timeout) to check all workers/stop.
class SupervisorThread:
def __init__(self, interval, workers): self.interval = interval self._workers = workers self._event = threading.Event()
def _run(self): while not self._event.is_set(): for worker in self._workers[:]: if worker.is_stuck(): worker.cancel() self._event.wait(self.interval) Line 120: Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit.
Francesco Romani has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(5 comments)
Added the missing answer. Will amend the docs do describe how this code is supposed to be used and what it is supposed to replace, so we (myself included) can have a clearer vision of the requirements.
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 79: """ Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue
Why do you need priorityqueue or scheduler?
To implement the periodic tasks.
The worker wakes up periodically and check if 1. there is new work to do? 2. if so, is the work due now or in the future?
This approach let the sampling code just submit once a periodic task to update the Vm stats, and let the pool handle the rest.
I think the whole periodic task thing and how this code is supposed to be used in sampling deserves some more detailed description, which I'm going to add to http://gerrit.ovirt.org/#/c/29190 Line 84: self._stop = threading.Event() Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name)
Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event()
This allows stopping immediately - but usually you want to stop when work i
Your solution to stop threads is more general and nicer, so I'll switch to it.
To cancel stuck thread is harder. We'll need a connection pool, or one connection per worker thread - not sure which solution is cleaner given our needs here.
We'll also need to make sure it is safe and supported for a thread to e.g. close the connection of a different thread. Not sure, need to check with libvirt. Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88:
Line 95: """ Line 96: execute a single try of work processing. Line 97: """ Line 98: with self._pull() as item: Line 99: if item is not None:
Why do you want to support None?
It is a relic due to the automatic queue reinjection feature implemented in the WorkQueue class, a feature than I now think just adds complexity with little if any benefit. To make a long story short: will remove and simplify. Line 100: self._do_work(item) Line 101: Line 102: def stop(self): Line 103: """
Line 115: yield item Line 116: except schedqueue.NotYet: Line 117: yield None Line 118: except schedqueue.Empty: Line 119: yield None
I don't understand why you need to inject event into the queue. I would imp
Nice supervisor code, I'll try to come up with something like that (and again, simplify things deeply).
For the events: I reinject them to implement the periodicity. Let's consider a trivial example: sample CPU stats for each VM. New code just submits a request to the pool in the form (interval, callable).
The pool knows it has to run the callable every <interval> seconds, and that preserve the current semantics in the VmStatsThread, which issue the underlying libvirt call every <interval> seconds.
But we want to get rid of these VmStatsThread, so we need either periodic tasks, so the pool automatically renews them, or some other code has to issue sampling calls periodically. Could be a scheduler thread for example, but I don't really like it.
I felt that periodic tasks could be a useful and meaningful addition, and this is the reason why I choose this route. Surely it can be greatly simplified.
However: I'll add some notes describing with more detail and references the case for periodic tasks and task reinjection, so everyone has a clear vision of the requirement and the code to be replaced, and we can then debate which is the best solution. Line 120: Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit.
Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit. Line 124: """ Line 125: tag, work, args, kwargs = item
work is a function or callable - why not call use an obvious name?
No good reason. Will change. Line 126: with self.track_time(tag): Line 127: try: Line 128: work(*args, **kwargs) Line 129: except Exception:
Michal Skrivanek has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 67: self._working_on = tag Line 68: self._work_started_at = self._timefunc() Line 69: yield Line 70: self._work_started_at = None Line 71: self._working_on = None
I'm not sure that detecting stuck thread by tracking time will work. Differ
I wouldn't necessarily consider different class of tasks in one pool.
why is libvirt connection pool a thing we need? I actually don't think it is a solution to push out the problem to libvirt. Eventually this leads to the current state.
The basic assumption/motivation is that we have tasks which can't be cancelled and we want to quarantine them. Line 72: Line 73: Line 74: class Worker(TimeTrackingThread): Line 75: """
Michal Skrivanek has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event()
Your solution to stop threads is more general and nicer, so I'll switch to
I don't think there is a solution to cancel a stuck thread. IMO the only way is to flag it to fail once it wakes up and quarantine the operation/task/VM Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88:
Nir Soffer has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(3 comments)
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 79: """ Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue
To implement the periodic tasks.
Scheduling is not the responsibility of a threadpool. You are mixing unrelated stuff.
If you like to scheduled stuff, add a scheduler that wakes up when the next job is ready and run or submit it to a thread pool dedicated to running such tasks.
Where do we need such scheduling in vdsm? Line 84: self._stop = threading.Event() Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name)
Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event()
Your solution to stop threads is more general and nicer, so I'll switch to
Connection pool is probably what we need, and each connection need a thread since you cannot submit multiple requests on the same connection without getting a response for the previous - right?
This is similar to database connection pool or http connection pool. Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88:
Line 115: yield item Line 116: except schedqueue.NotYet: Line 117: yield None Line 118: except schedqueue.Empty: Line 119: yield None
Nice supervisor code, I'll try to come up with something like that (and aga
ok, we need priodic tasks - so lets design a scheduler for these - not a thread pool.
Since these tasks are periodic, we have a very clear timeout for them - if a call that must run every 15 seconds did not run yet because it is waiting in the queue, or because libvirt connection is stuck, we should cancel this call.
Otherwise we will have huge queue of tasks that grows and grows without limit, giving irrelevant results (task ran minutes after its scheduled run time).
This will also make sure that we will never consume too much cpu - if we cannot get task to run because of slow response time and too many task to run, we will run only some of the tasks.
I built similar schedulers, using a heapq (https://docs.python.org/2.6/library/heapq.html) and poll, polling until the next scheduled call should run. Using such scheduler, we can schedule a cancel operation on a tak when we start the task, and we don't need to supervise the running task.
Again, this is not related to a threadpool. A threapool can be used to run scheduled tasks concurrently. Line 120: Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit.
Michal Skrivanek has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event()
Connection pool is probably what we need, and each connection need a thread
by connection pool you mean libvirt connections pool? that's exactly what i don't want to have. It's not a solution I'm looking for to just push out the problem to libvirt and exhaust their worker pool.
as far as stat queries are concerned…if the per-VM retrieval task gets stuck, I just want to skip that one VM stat retrieval till it recovers/cancels. I do not want to allocate another connection Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88:
Nir Soffer has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(3 comments)
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 67: self._working_on = tag Line 68: self._work_started_at = self._timefunc() Line 69: yield Line 70: self._work_started_at = None Line 71: self._working_on = None
I wouldn't necessarily consider different class of tasks in one pool.
There no point in keeping 10s of libvirt connections stuck. We should close stuck connections and open new connections instead.
Why do you think that connection pool will push the problem to libvirt? we can limit the number of connections and be nice to libvirt. Line 72: Line 73: Line 74: class Worker(TimeTrackingThread): Line 75: """
Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event()
by connection pool you mean libvirt connections pool?
Miachl, we don't want thread per vm, which is not scale-able as you just reported in the devel list. One stuck connection degrade the performance of the whole pool. So closing such connection si the nicest thing we can do, unless thats breaks libvirt badly :-) Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88:
Line 80: def __init__(self, work_queue): Line 81: super(Worker, self).__init__() Line 82: self.daemon = True Line 83: self._work_queue = work_queue Line 84: self._stop = threading.Event()
I don't think there is a solution to cancel a stuck thread. IMO the only wa
Michal, do you want to stop stats collections temporary for a vm that fails?
If so, then vm object should keep its stats health, and schedule new stats calls according to its health - we don't need periodic tasks in this case. Each stat call can schedule the next one, preventing flooding in case libvirt does not respond well for queries on this vm. Line 85: Line 86: def run(self): Line 87: logging.info('worker %s starting', self.name) Line 88:
Francesco Romani has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
patch set 9: rebased due to changes in parent patch.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 9:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/9937/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10722/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/66/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/99/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/78/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/10879/ : SUCCESS
Michal Skrivanek has posted comments on this change.
Change subject: threadpool: add a thread pool with watchdog ......................................................................
Patch Set 8:
(2 comments)
http://gerrit.ovirt.org/#/c/29191/8/lib/threadpool/worker.py File lib/threadpool/worker.py:
Line 67: self._working_on = tag Line 68: self._work_started_at = self._timefunc() Line 69: yield Line 70: self._work_started_at = None Line 71: self._working_on = None
There no point in keeping 10s of libvirt connections stuck. We should close
we should try to "close", yes, that may mean more harsh recovery..like restart libvirt and restart vdsm. Which is fine…I'd not try to do it in the first iteration, though. Line 72: Line 73: Line 74: class Worker(TimeTrackingThread): Line 75: """
Line 115: yield item Line 116: except schedqueue.NotYet: Line 117: yield None Line 118: except schedqueue.Empty: Line 119: yield None
ok, we need priodic tasks - so lets design a scheduler for these - not a th
ok. well, we need concrete implementation…for generic threadpool concurrent.futures looks like a good fit, we can build on top of that. After all, the watchdog/watcher is where all the logic is going to be Line 120: Line 121: def _do_work(self, item): Line 122: """ Line 123: process a single work unit.
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 10:
trimmed away all the bloat and reduced the concept to its barebones. Refacored the tests (test coverage is >= 95% according to nose). Added docstrings.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 10:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10205/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/10990/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/106... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/137... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/116/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11147/ : FAILURE
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 10: Code-Review-1
kept for context until we settle down for the final solution. No longer the preferred way to go.
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 10: -Code-Review
discard previous comment
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 10:
(27 comments)
Nice, can be simplified and be more correct.
http://gerrit.ovirt.org/#/c/29191/10/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 36: pass Line 37: Line 38: Line 39: def _COLLECTOR(result): Line 40: pass Can you move private stuff like this to the end of the file? It is easier to understand how to use this when you don't see the public parts first. Line 41: Line 42: Line 43: class NoAvailableWorker(Exception): Line 44: """
Line 45: No available worker to carry the task. Line 46: """ Line 47: Line 48: Line 49: class PoolNotRunning(Exception): Lets go with your choice and name it ExecutorNotRunning Line 50: """ Line 51: Pool not yet started or shutting down. Line 52: """ Line 53:
Line 59: """ Line 60: Line 61: _log = logging.getLogger("Executor") Line 62: Line 63: def __init__(self, num_workers=10): I think that "workers" or "size" is much more readable then the cryptic num_workers. If you want to be more verbose, maybe "worker_count". Line 64: self._num_workers = num_workers Line 65: self._tasks = Queue.Queue() Line 66: self._workers = [] Line 67: self._resize_lock = threading.Lock()
Line 61: _log = logging.getLogger("Executor") Line 62: Line 63: def __init__(self, num_workers=10): Line 64: self._num_workers = num_workers Line 65: self._tasks = Queue.Queue() We need to limit the size - you don't want to hold here 1000's of tasks.
When the tasks queue is full, you probably want to raise, so the caller can handle this without blocking. Line 66: self._workers = [] Line 67: self._resize_lock = threading.Lock() Line 68: self._running = False Line 69: self._evictions = 0 # for debug purposes only
Line 63: def __init__(self, num_workers=10): Line 64: self._num_workers = num_workers Line 65: self._tasks = Queue.Queue() Line 66: self._workers = [] Line 67: self._resize_lock = threading.Lock() I would call this _lock, since you need to protect more then the workers list. Line 68: self._running = False Line 69: self._evictions = 0 # for debug purposes only Line 70: Line 71: def start(self):
Line 68: self._running = False Line 69: self._evictions = 0 # for debug purposes only Line 70: Line 71: def start(self): Line 72: with self._resize_lock: You should raise if executor is started multiple times. Line 73: for _ in range(self._num_workers): Line 74: self._add_worker() Line 75: self._running = True Line 76:
Line 74: self._add_worker() Line 75: self._running = True Line 76: Line 77: def stop(self): Line 78: self._running = False Must be inside the lock. Line 79: for worker in self._workers: Line 80: self._tasks.put((_STOP, (), {})) Line 81: with self._resize_lock: Line 82: for worker in self._workers:
Line 79: for worker in self._workers: Line 80: self._tasks.put((_STOP, (), {})) Line 81: with self._resize_lock: Line 82: for worker in self._workers: Line 83: self._log.debug('terminating worker: %s', worker.name) This is not correct, you are not terminating the worker here. Maybe: ("Worker %s stopped", worker.name) Line 84: worker.join() Line 85: del worker Line 86: Line 87: @property
Line 80: self._tasks.put((_STOP, (), {})) Line 81: with self._resize_lock: Line 82: for worker in self._workers: Line 83: self._log.debug('terminating worker: %s', worker.name) Line 84: worker.join() This can take lot of time - should be done only if the caller specified wait=True Line 85: del worker Line 86: Line 87: @property Line 88: def size(self):
Line 84: worker.join() Line 85: del worker Line 86: Line 87: @property Line 88: def size(self): If you go with "size" here, then the __init__ argument should be also "size".
But do we need this at all? checking the size is racy, better not report it. Line 89: """ Line 90: returns the actual size of the worker pool. Line 91: """ Line 92: with self._resize_lock:
Line 91: """ Line 92: with self._resize_lock: Line 93: return len(self._workers) Line 94: Line 95: def send(self, task, *args, **kwargs): send? I don't understand this api.
I think we need one interface such as "dispatch" or "queue". send is very strange interface for thread pool.
I also prefer not to support args and kwargs, just accept a callable. If the user want to keep additional state, it should send a callable keeping this state.
We can always add *args or even **kwargs if needed, lets start with simpler version. Line 96: """ Line 97: send a new task to the pool. Line 98: Line 99: The task may be any callable. extra arguments are
Line 106: See doc(BackgroundTask.discard) Line 107: """ Line 108: return self.submit(_COLLECTOR, task, *args, **kwargs) Line 109: Line 110: def submit(self, collector, task, *args, **kwargs): submit? I don't understand this api. Line 111: """ Line 112: submit a task to the pool, with a collector function. Line 113: Line 114: The collector function must be a callable which
Line 122: The collector will be called only if the task is not Line 123: discarded. Line 124: See doc(BackgroundTask.discard) Line 125: """ Line 126: if not self._running: Should be protected by the lock Line 127: raise PoolNotRunning Line 128: if not self._workers: Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector)
Line 124: See doc(BackgroundTask.discard) Line 125: """ Line 126: if not self._running: Line 127: raise PoolNotRunning Line 128: if not self._workers: Must be protected by the lock Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task
Line 125: """ Line 126: if not self._running: Line 127: raise PoolNotRunning Line 128: if not self._workers: Line 129: raise NoAvailableWorker We don't need to check for no workers - the executor should keep this list full, and we don't care if in some point in time it is empty, since the executor will fill it back soon. Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task Line 133:
Line 128: if not self._workers: Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task Keeping tuples of (task, args, kwargs) means that any object references from task (e.g. self for a method), args or kwargs, will never be collected until the task is executed. Since wrap the task with the BackgroundTask, I think you should put the args and kwargs inside this object, and put it in the queue.
Then running or discarding can be localized into the BackgroundTask object. Line 133: Line 134: def _run(self): Line 135: self._log.info('worker started') Line 136: while True:
Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task Line 133: Must be protected with @traceback Line 134: def _run(self): Line 135: self._log.info('worker started') Line 136: while True: Line 137: task, args, kwargs = self._tasks.get()
Line 134: def _run(self): Line 135: self._log.info('worker started') Line 136: while True: Line 137: task, args, kwargs = self._tasks.get() Line 138: if task is _STOP: This should be:
task = self._tasks.get() Line 139: break Line 140: try: Line 141: with running(task): Line 142: task(*args, **kwargs)
Line 138: if task is _STOP: Line 139: break Line 140: try: Line 141: with running(task): Line 142: task(*args, **kwargs) Should be:
task.execute() Line 143: except Exception: Line 144: self._log.exception('unhandled exception for task') Line 145: if task.discarded: Line 146: self._log.debug('worker aborting')
Line 140: try: Line 141: with running(task): Line 142: task(*args, **kwargs) Line 143: except Exception: Line 144: self._log.exception('unhandled exception for task') unhandled -> Unhandled Line 145: if task.discarded: Line 146: self._log.debug('worker aborting') Line 147: break Line 148: self._log.info('worker stopped')
Line 141: with running(task): Line 142: task(*args, **kwargs) Line 143: except Exception: Line 144: self._log.exception('unhandled exception for task') Line 145: if task.discarded: Nice and simple. Line 146: self._log.debug('worker aborting') Line 147: break Line 148: self._log.info('worker stopped') Line 149:
Line 142: task(*args, **kwargs) Line 143: except Exception: Line 144: self._log.exception('unhandled exception for task') Line 145: if task.discarded: Line 146: self._log.debug('worker aborting') Lets be more clear: "Worker stopping after task was discarded" Line 147: break Line 148: self._log.info('worker stopped') Line 149: Line 150: def remove_worker(self, worker):
Line 144: self._log.exception('unhandled exception for task') Line 145: if task.discarded: Line 146: self._log.debug('worker aborting') Line 147: break Line 148: self._log.info('worker stopped') I think we should handle removal of workers here:
with self._lock: me = threading.current_thread() self._workers.remove(me)
This can never fail since this is the only place removing from the workers list.
If we do this, size does not report the number of available threads, but you can do something like this:
class BackgroundTask:
def discard(self): self._thread.discarded = True self._executor._add_worker()
And if you like to query the state of the executor:
def available_workers(self): with self._lock: return len(t for t in self.workers if not t.discarded)
def discarded_workers(self): with self._lock: return len(t for t in self.workers if t.discarded)
Since number of workers is small, this is fast enough.
This also allows to wait for all threads, including stuck threads, but we probably don't want this
Anyway, having all threads, both active and stuck/discarded in same list will help to understand how this behaves. Line 149: Line 150: def remove_worker(self, worker): Line 151: "to be used from BackgroundTask only." Line 152: with self._resize_lock:
Line 146: self._log.debug('worker aborting') Line 147: break Line 148: self._log.info('worker stopped') Line 149: Line 150: def remove_worker(self, worker): This is little ugly - people will use this eventually. Just make this _protected and use it from the background task. Line 151: "to be used from BackgroundTask only." Line 152: with self._resize_lock: Line 153: if not self._running: Line 154: return # no point in messing up things if shutting down
Line 166: worker = threading.Thread(target=self._run) Line 167: worker.daemon = True Line 168: worker.start() Line 169: except Exception: Line 170: self._log.exception('cannot create worker') Let not handle this here, and let the caller handle this error. Line 171: else: Line 172: self._workers.append(worker) Line 173: self._log.debug('worker %s joined the pool', worker.name) Line 174:
Line 186: def __init__(self, task, executor, collector): Line 187: self._task = task Line 188: self._executor = executor Line 189: self._collector = collector Line 190: self._valid = True I don't understand why we need a collector. The task should simply run, if it should call some other function with some value, this should be implemented by the task function.
I don't think we should provide such services like the current storage thread pool.
This will allow removal of the collector, valid and one of the executor methods to dispatch tasks. Line 191: self._thread = None Line 192: Line 193: @property Line 194: def in_progress(self):
Line 217: self._valid = False Line 218: if self.in_progress: Line 219: self._executor.remove_worker(self._thread) Line 220: Line 221: # runnable interface: internal usage only This is nice, but internal stuff should be _protected.
But we don't need this fancy running interface, because we can do:
def execute(self): self._thread = threading.current_thread() try: self._task(self._args, self._kwargs) finally: self._thread = None Line 222: Line 223: def start(self): Line 224: self._thread = threading.current_thread() Line 225:
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 10:
(25 comments)
http://gerrit.ovirt.org/#/c/29191/10/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 36: pass Line 37: Line 38: Line 39: def _COLLECTOR(result): Line 40: pass
Can you move private stuff like this to the end of the file? It is easier t
Sure, will have the public interface first. Line 41: Line 42: Line 43: class NoAvailableWorker(Exception): Line 44: """
Line 45: No available worker to carry the task. Line 46: """ Line 47: Line 48: Line 49: class PoolNotRunning(Exception):
Lets go with your choice and name it ExecutorNotRunning
Done Line 50: """ Line 51: Pool not yet started or shutting down. Line 52: """ Line 53:
Line 59: """ Line 60: Line 61: _log = logging.getLogger("Executor") Line 62: Line 63: def __init__(self, num_workers=10):
I think that "workers" or "size" is much more readable then the cryptic num
Picked 'workers' Line 64: self._num_workers = num_workers Line 65: self._tasks = Queue.Queue() Line 66: self._workers = [] Line 67: self._resize_lock = threading.Lock()
Line 61: _log = logging.getLogger("Executor") Line 62: Line 63: def __init__(self, num_workers=10): Line 64: self._num_workers = num_workers Line 65: self._tasks = Queue.Queue()
We need to limit the size - you don't want to hold here 1000's of tasks.
Done Line 66: self._workers = [] Line 67: self._resize_lock = threading.Lock() Line 68: self._running = False Line 69: self._evictions = 0 # for debug purposes only
Line 63: def __init__(self, num_workers=10): Line 64: self._num_workers = num_workers Line 65: self._tasks = Queue.Queue() Line 66: self._workers = [] Line 67: self._resize_lock = threading.Lock()
I would call this _lock, since you need to protect more then the workers li
Done Line 68: self._running = False Line 69: self._evictions = 0 # for debug purposes only Line 70: Line 71: def start(self):
Line 68: self._running = False Line 69: self._evictions = 0 # for debug purposes only Line 70: Line 71: def start(self): Line 72: with self._resize_lock:
You should raise if executor is started multiple times.
Done Line 73: for _ in range(self._num_workers): Line 74: self._add_worker() Line 75: self._running = True Line 76:
Line 74: self._add_worker() Line 75: self._running = True Line 76: Line 77: def stop(self): Line 78: self._running = False
Must be inside the lock.
Done Line 79: for worker in self._workers: Line 80: self._tasks.put((_STOP, (), {})) Line 81: with self._resize_lock: Line 82: for worker in self._workers:
Line 79: for worker in self._workers: Line 80: self._tasks.put((_STOP, (), {})) Line 81: with self._resize_lock: Line 82: for worker in self._workers: Line 83: self._log.debug('terminating worker: %s', worker.name)
This is not correct, you are not terminating the worker here. Maybe: ("Work
Picked 'Worker %s stopped' Line 84: worker.join() Line 85: del worker Line 86: Line 87: @property
Line 80: self._tasks.put((_STOP, (), {})) Line 81: with self._resize_lock: Line 82: for worker in self._workers: Line 83: self._log.debug('terminating worker: %s', worker.name) Line 84: worker.join()
This can take lot of time - should be done only if the caller specified wai
Done Line 85: del worker Line 86: Line 87: @property Line 88: def size(self):
Line 84: worker.join() Line 85: del worker Line 86: Line 87: @property Line 88: def size(self):
If you go with "size" here, then the __init__ argument should be also "size
No immediaten need, then removed. Line 89: """ Line 90: returns the actual size of the worker pool. Line 91: """ Line 92: with self._resize_lock:
Line 91: """ Line 92: with self._resize_lock: Line 93: return len(self._workers) Line 94: Line 95: def send(self, task, *args, **kwargs):
send? I don't understand this api.
Ok for the naming. Will use 'dispatch'. Agreed for *args and **kwargs: no immediate need, let's start without them. Line 96: """ Line 97: send a new task to the pool. Line 98: Line 99: The task may be any callable. extra arguments are
Line 106: See doc(BackgroundTask.discard) Line 107: """ Line 108: return self.submit(_COLLECTOR, task, *args, **kwargs) Line 109: Line 110: def submit(self, collector, task, *args, **kwargs):
submit? I don't understand this api.
Well, in retrospect yes, those are ugly names. I needed somehow similar names to dispatch a task with and without a collector callable. But let's just get rid of the collector until (if any) there is objective need for it. Line 111: """ Line 112: submit a task to the pool, with a collector function. Line 113: Line 114: The collector function must be a callable which
Line 122: The collector will be called only if the task is not Line 123: discarded. Line 124: See doc(BackgroundTask.discard) Line 125: """ Line 126: if not self._running:
Should be protected by the lock
Done Line 127: raise PoolNotRunning Line 128: if not self._workers: Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector)
Line 124: See doc(BackgroundTask.discard) Line 125: """ Line 126: if not self._running: Line 127: raise PoolNotRunning Line 128: if not self._workers:
Must be protected by the lock
Done Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task
Line 128: if not self._workers: Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task
Keeping tuples of (task, args, kwargs) means that any object references fro
Agreed to get rid of explicit args and kwargs and pack the state in the object Line 133: Line 134: def _run(self): Line 135: self._log.info('worker started') Line 136: while True:
Line 129: raise NoAvailableWorker Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task Line 133:
Must be protected with @traceback
Done Line 134: def _run(self): Line 135: self._log.info('worker started') Line 136: while True: Line 137: task, args, kwargs = self._tasks.get()
Line 134: def _run(self): Line 135: self._log.info('worker started') Line 136: while True: Line 137: task, args, kwargs = self._tasks.get() Line 138: if task is _STOP:
This should be:
Done Line 139: break Line 140: try: Line 141: with running(task): Line 142: task(*args, **kwargs)
Line 138: if task is _STOP: Line 139: break Line 140: try: Line 141: with running(task): Line 142: task(*args, **kwargs)
Should be:
Done Line 143: except Exception: Line 144: self._log.exception('unhandled exception for task') Line 145: if task.discarded: Line 146: self._log.debug('worker aborting')
Line 140: try: Line 141: with running(task): Line 142: task(*args, **kwargs) Line 143: except Exception: Line 144: self._log.exception('unhandled exception for task')
unhandled -> Unhandled
Done Line 145: if task.discarded: Line 146: self._log.debug('worker aborting') Line 147: break Line 148: self._log.info('worker stopped')
Line 142: task(*args, **kwargs) Line 143: except Exception: Line 144: self._log.exception('unhandled exception for task') Line 145: if task.discarded: Line 146: self._log.debug('worker aborting')
Lets be more clear: "Worker stopping after task was discarded"
Done Line 147: break Line 148: self._log.info('worker stopped') Line 149: Line 150: def remove_worker(self, worker):
Line 144: self._log.exception('unhandled exception for task') Line 145: if task.discarded: Line 146: self._log.debug('worker aborting') Line 147: break Line 148: self._log.info('worker stopped')
I think we should handle removal of workers here:
This will lead to better code, so I will do it. Line 149: Line 150: def remove_worker(self, worker): Line 151: "to be used from BackgroundTask only." Line 152: with self._resize_lock:
Line 146: self._log.debug('worker aborting') Line 147: break Line 148: self._log.info('worker stopped') Line 149: Line 150: def remove_worker(self, worker):
This is little ugly - people will use this eventually. Just make this _prot
Agreed and done Line 151: "to be used from BackgroundTask only." Line 152: with self._resize_lock: Line 153: if not self._running: Line 154: return # no point in messing up things if shutting down
Line 166: worker = threading.Thread(target=self._run) Line 167: worker.daemon = True Line 168: worker.start() Line 169: except Exception: Line 170: self._log.exception('cannot create worker')
Let not handle this here, and let the caller handle this error.
Done Line 171: else: Line 172: self._workers.append(worker) Line 173: self._log.debug('worker %s joined the pool', worker.name) Line 174:
Line 186: def __init__(self, task, executor, collector): Line 187: self._task = task Line 188: self._executor = executor Line 189: self._collector = collector Line 190: self._valid = True
I don't understand why we need a collector. The task should simply run, if
No immediate or obvious need for the collector, just a maybe-nice-to-have feature. Thus removed and simplified the code. Line 191: self._thread = None Line 192: Line 193: @property Line 194: def in_progress(self):
Line 217: self._valid = False Line 218: if self.in_progress: Line 219: self._executor.remove_worker(self._thread) Line 220: Line 221: # runnable interface: internal usage only
This is nice, but internal stuff should be _protected.
Agreed and removed. Line 222: Line 223: def start(self): Line 224: self._thread = threading.current_thread() Line 225:
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 10:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/10/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 125: """ Line 126: if not self._running: Line 127: raise PoolNotRunning Line 128: if not self._workers: Line 129: raise NoAvailableWorker
We don't need to check for no workers - the executor should keep this list
Done Line 130: task = BackgroundTask(task, self, collector) Line 131: self._tasks.put((task, args, kwargs)) Line 132: return task Line 133:
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 11:
addressed Nir's comments. Will do additional verification for MT safety.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 11:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10317/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11102/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/121... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/152... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/131/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11259/ : FAILURE
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 11:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/11/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 114: break Line 115: self._log.info('worker stopped') Line 116: if task.discarded: Line 117: self._remove_worker(threading.current_thread()) Line 118: # else no action needed: stop(wait=True) will take care. Need to make sure there is no deadlock here if BackgroundTask.discard() is called while the Executor is stop(wait=True)'ing Line 119: Line 120: def _remove_worker(self, worker): Line 121: with self._lock: Line 122: self._workers.remove(worker)
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 11:
(21 comments)
I think that this is the right direction. To discard tasks, the code which submit the should schedule a task.discard() call using the scheduler. The calling code should be able to tell the right timeout for this.
There are various issues in the implementation. See the inline comments.
To shorten the review cycle, I posted a modified version that show how all the issue can be solved: http://pastebin.com/xevn4b28
http://gerrit.ovirt.org/#/c/29191/11/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 28: import logging Line 29: import Queue Line 30: import threading Line 31: Line 32: import utils We can use now relative imports:
from . import utils Line 33: Line 34: Line 35: class ExecutorNotRunning(Exception): Line 36: """
Line 31: Line 32: import utils Line 33: Line 34: Line 35: class ExecutorNotRunning(Exception): "executor.ExecutorNotRunning" repeat the "Executor" word for no reason. We have namespaces in Python, so there is no need to repeat stuff. Line 36: """ Line 37: Executor not yet started or shutting down. Line 38: """ Line 39:
Line 37: Executor not yet started or shutting down. Line 38: """ Line 39: Line 40: Line 41: class ExecutorAlreadyStarted(Exception): "executor.ExecutorAllreadyStarted" repeat the "Executor" word for no reason. Line 42: """ Line 43: Executor started multiple times. Line 44: """ Line 45:
Line 51: """ Line 52: Line 53: _log = logging.getLogger("Executor") Line 54: Line 55: def __init__(self, workers=10, max_tasks=1000): I don't think we need defaults, since we don't know how the executor is going to be used. The caller must specify this. Line 56: self._workers_count = workers Line 57: self._tasks = Queue.Queue(max_tasks) Line 58: self._workers = [] Line 59: self._lock = threading.Lock()
Line 54: Line 55: def __init__(self, workers=10, max_tasks=1000): Line 56: self._workers_count = workers Line 57: self._tasks = Queue.Queue(max_tasks) Line 58: self._workers = [] Lets keep workers in a set(), make it easier to remove them when needed. Line 59: self._lock = threading.Lock() Line 60: self._running = False Line 61: self._discards = 0 # for debug purposes only Line 62:
Line 57: self._tasks = Queue.Queue(max_tasks) Line 58: self._workers = [] Line 59: self._lock = threading.Lock() Line 60: self._running = False Line 61: self._discards = 0 # for debug purposes only I don't think we need this, logging each discard operation is enough. Line 62: Line 63: def start(self): Line 64: with self._lock: Line 65: if self._running:
Line 65: if self._running: Line 66: raise ExecutorAlreadyStarted Line 67: for _ in range(self._workers_count): Line 68: self._add_worker() Line 69: self._running = True Lets put self._running before starting the threads, which may use this value. Line 70: Line 71: def stop(self, wait=True): Line 72: for worker in self._workers: Line 73: self._tasks.put(_STOP)
Line 69: self._running = True Line 70: Line 71: def stop(self, wait=True): Line 72: for worker in self._workers: Line 73: self._tasks.put(_STOP) This may block if tasks queue is full, so this is not a good way to stop the threads.
This will all threads *after* pending tasks are executed, which can cause threads to stuck, and this method will never return when using wait=True.
I think that we like to stop all waiting threads *now*, and drop pending tasks.
To do this, we need to clear the queue, and signal all threads to exit. This is not possible when using the builtin Queue. Queue contains many features which we don't need (task_done, join), but not what we need (clear). Line 74: with self._lock: Line 75: self._running = False Line 76: if wait: Line 77: for worker in self._workers:
Line 72: for worker in self._workers: Line 73: self._tasks.put(_STOP) Line 74: with self._lock: Line 75: self._running = False Line 76: if wait: This will deadlock because you hold the lock when exiting threads may try to acquire it.
Also this wait is too dangerous, since a thread which is running a task may get stuck, and thread which is already running a task may (practically) never exit.
We like to wait only for threads which are not executing any task. Line 77: for worker in self._workers: Line 78: self._log.debug('worker %s stopped', worker.name) Line 79: worker.join() Line 80: del worker
Line 94: with self._lock: Line 95: if not self._running: Line 96: raise ExecutorNotRunning Line 97: task = BackgroundTask(task, self) Line 98: self._tasks.put(task) This may raise as you released the lock above, so another thread may have just stopped the executor.
Also, adding task may block if the queue is full. This breaks other code that assume that dispatch is async - we never want to do this.
We should use here put_nowait() to ensure that this never blocks. The caller of dispatch should handle the Queue.Full exception that this raises. Line 99: return task Line 100: Line 101: @utils.traceback() Line 102: def _run(self):
Line 108: try: Line 109: task.execute() Line 110: except Exception: Line 111: self._log.exception('Unhandled exception for task') Line 112: if task.discarded: This can be simplified by raising in task.execute() if task was discarded. Line 113: self._log.debug('Worker stopping after task was discarded') Line 114: break Line 115: self._log.info('worker stopped') Line 116: if task.discarded:
Line 112: if task.discarded: Line 113: self._log.debug('Worker stopping after task was discarded') Line 114: break Line 115: self._log.info('worker stopped') Line 116: if task.discarded: We don't need this check, stop does not remove workers from the workers list. Line 117: self._remove_worker(threading.current_thread()) Line 118: # else no action needed: stop(wait=True) will take care. Line 119: Line 120: def _remove_worker(self, worker):
Line 122: self._workers.remove(worker) Line 123: self._log.debug('worker %s left the executor', worker.name) Line 124: Line 125: def _add_worker(self): Line 126: worker = threading.Thread(target=self._run) Name this thread - to make it easy to understand the logs, I wouuld keep a counter and call the threads "Executor-N" Line 127: worker.daemon = True Line 128: worker.start() Line 129: self._workers.append(worker) Line 130: self._log.debug('worker %s joined the executor', worker.name)
Line 131: Line 132: def _replace_worker(self): Line 133: with self._lock: Line 134: self._discards += 1 Line 135: self._add_worker() This should not add a worker if executor was stopped. Line 136: Line 137: Line 138: class BackgroundTask(object): Line 139: """
Line 134: self._discards += 1 Line 135: self._add_worker() Line 136: Line 137: Line 138: class BackgroundTask(object): More specific name would be ExecutorTask. Line 139: """ Line 140: Represents a task submitted to the Executor. Line 141: The client can Line 142: - verify if a task is in progress or not
Line 147: """ Line 148: def __init__(self, task, executor): Line 149: self._task = task Line 150: self._executor = executor Line 151: self._thread = None We don't need the actual thread, keeping a boolean is enough. Line 152: self._discarded = False Line 153: Line 154: @property Line 155: def discarded(self):
Line 159: def in_progress(self): Line 160: """ Line 161: returns True if this task is in progress, False otherwise. Line 162: """ Line 163: return self._thread is not None This may be a task waiting on the queue because all workers are busy, or a task that was already executed.
Why do we need this racy check? Line 164: Line 165: def discard(self): Line 166: """ Line 167: This is to be called once the task is detected as stuck.
Line 172: It is not possible to interrupt a stuck task, thus Line 173: any side effect will still take place. Line 174: """ Line 175: self._discarded = True Line 176: self._executor._replace_worker() This is racy - if a thread was just executing this task, discarding may be too late, so the thread will continue, and we will add another thread to the pool for no reason.
For example:
1. thread 1 executing task 2. thread 2 call discard 3. thread 1 finish task 4. thread 2 set self._discard and invoke _replace_worker
Also, calling an executor protected method is little ugly. Since we don't care about the executor object, but only about _replace_worker, we can accept the method instead of the object.
Finally, discarding same task twice will add 2 new workers instead of one. Line 177: Line 178: # task interface Line 179: Line 180: def execute(self):
Line 176: self._executor._replace_worker() Line 177: Line 178: # task interface Line 179: Line 180: def execute(self): Raising if this task is already executing will be nice - this is impossible and should raise AssertionError. Line 181: self._thread = threading.current_thread() Line 182: try: Line 183: self._task() Line 184: finally:
Line 181: self._thread = threading.current_thread() Line 182: try: Line 183: self._task() Line 184: finally: Line 185: self._thread = None This must release the reference to the task, since this reference prevent garbage collection of the task until the background task is collected.
Also keeping reference to the executor make it harder for Python to collect the executor (executor -> tasks -> task -> executor).
We can simplify the worker thread logic by raising here if task was discarded while executing. Line 186: Line 187:
Line 184: finally: Line 185: self._thread = None Line 186: Line 187: Line 188: _STOP = BackgroundTask(None, None) This should be object() - nobody get this back and expect the BackgroundTask interface, and we want it to break if someone try to execute it.
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 11:
I checked integrating this in the sampling patch (http://gerrit.ovirt.org/29980), and it seems that passing the responsibility for discarding tasks on the caller is not the right solution.
The caller of this executor should not care about maintaining available workers in the executor. This should be handled by the executor. Otherwise we will have to duplicate this code in all the clients.
To keep this simple, we can add an optional timeout argument to dispatch:
def dispatch(self, task, timeout=None): ...
The timeout can mean, this task is expected to finish in timeout seconds, and if not, you can consider this task as stuck, and do what you need to do about that. So the caller has full control, but does not have to care about how timed out task are handled.
To handle the timeout, we can schedule the BackgroundThread.discard() call using a scheduler. If the timeout expired while the call is running, the worker thread will be discarded. When the call finish.
This is an example how we can implement this: http://pastebin.com/iMTN8wkA
Since the caller does not care any more about discarding, we can simplify the interface and return nothing from dispatch().
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 11:
(14 comments)
initial response
http://gerrit.ovirt.org/#/c/29191/11/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 28: import logging Line 29: import Queue Line 30: import threading Line 31: Line 32: import utils
We can use now relative imports:
Done Line 33: Line 34: Line 35: class ExecutorNotRunning(Exception): Line 36: """
Line 31: Line 32: import utils Line 33: Line 34: Line 35: class ExecutorNotRunning(Exception):
"executor.ExecutorNotRunning" repeat the "Executor" word for no reason. We
Sure. removed the repetition. Line 36: """ Line 37: Executor not yet started or shutting down. Line 38: """ Line 39:
Line 37: Executor not yet started or shutting down. Line 38: """ Line 39: Line 40: Line 41: class ExecutorAlreadyStarted(Exception):
"executor.ExecutorAllreadyStarted" repeat the "Executor" word for no reason
As above Line 42: """ Line 43: Executor started multiple times. Line 44: """ Line 45:
Line 51: """ Line 52: Line 53: _log = logging.getLogger("Executor") Line 54: Line 55: def __init__(self, workers=10, max_tasks=1000):
I don't think we need defaults, since we don't know how the executor is goi
Good point, removed. Line 56: self._workers_count = workers Line 57: self._tasks = Queue.Queue(max_tasks) Line 58: self._workers = [] Line 59: self._lock = threading.Lock()
Line 54: Line 55: def __init__(self, workers=10, max_tasks=1000): Line 56: self._workers_count = workers Line 57: self._tasks = Queue.Queue(max_tasks) Line 58: self._workers = []
Lets keep workers in a set(), make it easier to remove them when needed.
Done Line 59: self._lock = threading.Lock() Line 60: self._running = False Line 61: self._discards = 0 # for debug purposes only Line 62:
Line 65: if self._running: Line 66: raise ExecutorAlreadyStarted Line 67: for _ in range(self._workers_count): Line 68: self._add_worker() Line 69: self._running = True
Lets put self._running before starting the threads, which may use this valu
Done Line 70: Line 71: def stop(self, wait=True): Line 72: for worker in self._workers: Line 73: self._tasks.put(_STOP)
Line 69: self._running = True Line 70: Line 71: def stop(self, wait=True): Line 72: for worker in self._workers: Line 73: self._tasks.put(_STOP)
This may block if tasks queue is full, so this is not a good way to stop th
Agreed. Will work on this. Line 74: with self._lock: Line 75: self._running = False Line 76: if wait: Line 77: for worker in self._workers:
Line 72: for worker in self._workers: Line 73: self._tasks.put(_STOP) Line 74: with self._lock: Line 75: self._running = False Line 76: if wait:
This will deadlock because you hold the lock when exiting threads may try t
Will introduce a per-thread 'discarded' flag and will wait only for not-discarded threads as you suggested in a past comment.
Will also rearchitect the stop() to overcome this dangerous behaviour. Line 77: for worker in self._workers: Line 78: self._log.debug('worker %s stopped', worker.name) Line 79: worker.join() Line 80: del worker
Line 94: with self._lock: Line 95: if not self._running: Line 96: raise ExecutorNotRunning Line 97: task = BackgroundTask(task, self) Line 98: self._tasks.put(task)
This may raise as you released the lock above, so another thread may have j
Will move inside the lock and switch to put_nowait Line 99: return task Line 100: Line 101: @utils.traceback() Line 102: def _run(self):
Line 108: try: Line 109: task.execute() Line 110: except Exception: Line 111: self._log.exception('Unhandled exception for task') Line 112: if task.discarded:
This can be simplified by raising in task.execute() if task was discarded.
Good point, will change. Line 113: self._log.debug('Worker stopping after task was discarded') Line 114: break Line 115: self._log.info('worker stopped') Line 116: if task.discarded:
Line 122: self._workers.remove(worker) Line 123: self._log.debug('worker %s left the executor', worker.name) Line 124: Line 125: def _add_worker(self): Line 126: worker = threading.Thread(target=self._run)
Name this thread - to make it easy to understand the logs, I wouuld keep a
Done Line 127: worker.daemon = True Line 128: worker.start() Line 129: self._workers.append(worker) Line 130: self._log.debug('worker %s joined the executor', worker.name)
Line 134: self._discards += 1 Line 135: self._add_worker() Line 136: Line 137: Line 138: class BackgroundTask(object):
More specific name would be ExecutorTask.
Done Line 139: """ Line 140: Represents a task submitted to the Executor. Line 141: The client can Line 142: - verify if a task is in progress or not
Line 159: def in_progress(self): Line 160: """ Line 161: returns True if this task is in progress, False otherwise. Line 162: """ Line 163: return self._thread is not None
This may be a task waiting on the queue because all workers are busy, or a
No immediate need, will remove Line 164: Line 165: def discard(self): Line 166: """ Line 167: This is to be called once the task is detected as stuck.
Line 184: finally: Line 185: self._thread = None Line 186: Line 187: Line 188: _STOP = BackgroundTask(None, None)
This should be object() - nobody get this back and expect the BackgroundTas
Done
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 12:
Rebased on top of Nir's patches. Addressed most of the comments from Nir, more to come.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 12:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10353/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11138/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/133... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/164... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/143/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11296/ : FAILURE
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 12:
(5 comments)
I think this should be rebased on the schedule patch, under the sampling patch, so we can use this executor in the sampling patch.
http://gerrit.ovirt.org/#/c/29191/12/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 52: self._tasks = _Queue(max_tasks) Line 53: self._workers = set() Line 54: self._cond = threading.Condition(threading.Lock()) Line 55: self._running = False Line 56: self._scheduler = scheduler Can you initialize the object instance variables in the same order of the arguments, and do this before any other thing?
While this does not make any difference in the behavior, It makes it easier to understand the code. Line 57: self._worker_id = 0 Line 58: Line 59: def start(self): Line 60: self._log.debug('Starting executor')
Line 101: worker.busy = False Line 102: self._log.debug('Starting worker %s' % worker.name) Line 103: worker.start() Line 104: self._workers.add(worker) Line 105: self._workers.add(worker) I think that one of these is enough :-) Line 106: Line 107: def _task_discarded(self): Line 108: with self._cond: Line 109: if self._running:
Line 186: on_task_discarded = self._on_task_discarded Line 187: on_task_discarded() Line 188: Line 189: Line 190: class _Queue(object): Lets call it _TaskQueue, or rename the _max_tasks, and _tasks to more generic names.
Since this is not really a generic utility, making it more specific seems more correct. Line 191: """Queue.Queue doesn't support the clear() operation. Line 192: All the locking must be provided by the caller.""" Line 193: Line 194: def __init__(self, max_tasks):
Line 194: def __init__(self, max_tasks): Line 195: self._max_tasks = max_tasks Line 196: self._tasks = collections.deque() Line 197: Line 198: @property This does not makes sense as a property. Line 199: def empty(self): Line 200: return len(self._tasks) == 0 Line 201: Line 202: def put(self, task):
Line 207: def get(self): Line 208: return self._tasks.pop() Line 209: Line 210: def clear(self): Line 211: self._tasks.clear() Nice! I ignored this direction because of the locking issues, but without locking this makes the code much nicer. Line 212: Line 213: Line 214: class _ExecutorStopped(Exception): Line 215: pass
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 13:
Rebased on the scheduler patch, instead of the sampling patch, so I can use this in the sampling patch.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 13:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10371/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11156/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/135... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/166... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/145/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11313/ : FAILURE
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 14:
Rebased on scheduler patch after it was modified. No code change.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 14:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10378/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11163/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/137... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/168... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/147/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11320/ : FAILURE
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 15:
Addressed Nir's comments in patchset #12.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 15:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10381/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11166/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/140... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/171... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/150/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11324/ : FAILURE
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 16:
squashed 30742 and 30743 from nsoffer (thanks again Nir); 30744 and 3745 are the next in line. fixed call to schedule()
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 16:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10472/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11257/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/159... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/190... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/169/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11414/ : SUCCESS
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 17:
squashed 30745 from nsoffer (thanks Nir)
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 17:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10477/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11262/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/161... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/192... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/171/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11420/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 18:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/193/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10610/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/183... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11552/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/214... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11395/ : SUCCESS
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 18:
Rebase as http://gerrit.ovirt.org/29607 was modified.
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 18:
rebased
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 19:
rebased (discard previous comment)
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 19:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/214/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10706/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/204... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11648/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/236... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11491/ : SUCCESS
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 20:
now includes http://gerrit.ovirt.org/#/c/30744/ (thanks Nir)
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 21:
now includes http://gerrit.ovirt.org/#/c/30986 (thanks Nir!)
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 20:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/216/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10714/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/206... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11656/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/238... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11499/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 21:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/217/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10718/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/207... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11660/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/239... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11503/ : SUCCESS
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 21:
(4 comments)
http://gerrit.ovirt.org/#/c/29191/21/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 69: with self._lock: Line 70: self._running = False Line 71: self._tasks.clear() Line 72: for i in range(self._workers_count): Line 73: self._tasks.put((_STOP, 0)) Lets use None here, which mean no timeout. It does not matter because the code should check for _STOP, but 0 simply does not make sense. Line 74: workers = tuple(self._workers) if wait else () Line 75: for worker in workers: Line 76: try: Line 77: worker.join()
Line 210: def __init__(self, max_tasks): Line 211: self._max_tasks = max_tasks Line 212: self._tasks = collections.deque() Line 213: # deque is thread safe - we can append and pop from both ends without Line 214: # additioanl locking. We need this condition only for waiting. Lets add a url to the docs about this: https://docs.python.org/2.6/library/queue.html#Queue.Full
Yes, this fact is documented in Queue docs instead of deque docs :-) Line 215: self._cond = threading.Condition(threading.Lock()) Line 216: self._waiters = 0 Line 217: Line 218: def put(self, task):
Line 222: if self._waiters > 0: Line 223: with self._cond: Line 224: self._cond.notify() Line 225: Line 226: def get(self): We have tiny race here:
1. thread 1 call get and find the queue empty 2. context switch to thread 2 3. thread 2 call put, adding a task to the queue 4. thread 2 check self._waiters == 0 and return 5. context switch to thread 1 6. thread 1 set waiters to 1 and wait on the condition
I think that we will never see this race during sampling, since the waiting thread will be woke up by the next message, and we have a constant stream of tasks. But as a general utility, we should not have such races. Line 227: while True: Line 228: try: Line 229: return self._tasks.popleft() Line 230: except IndexError:
Line 229: return self._tasks.popleft() Line 230: except IndexError: Line 231: with self._cond: Line 232: self._waiters += 1 Line 233: self._cond.wait() To avoid the race explained above:
if self._tasks.empty(): self._cond.wait()
So if another thread added a task before I bumped _waiters, I will not wait.
Hopefully I'm not wrong about this, avoiding unneeded locking is tricky. Line 234: self._waiters -= 1 Line 235: Line 236: def clear(self):
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 21:
(3 comments)
http://gerrit.ovirt.org/#/c/29191/21/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 69: with self._lock: Line 70: self._running = False Line 71: self._tasks.clear() Line 72: for i in range(self._workers_count): Line 73: self._tasks.put((_STOP, 0))
Lets use None here, which mean no timeout. It does not matter because the c
Done Line 74: workers = tuple(self._workers) if wait else () Line 75: for worker in workers: Line 76: try: Line 77: worker.join()
Line 210: def __init__(self, max_tasks): Line 211: self._max_tasks = max_tasks Line 212: self._tasks = collections.deque() Line 213: # deque is thread safe - we can append and pop from both ends without Line 214: # additioanl locking. We need this condition only for waiting.
Lets add a url to the docs about this:
Done, and corrected a small typo along the way. Line 215: self._cond = threading.Condition(threading.Lock()) Line 216: self._waiters = 0 Line 217: Line 218: def put(self, task):
Line 229: return self._tasks.popleft() Line 230: except IndexError: Line 231: with self._cond: Line 232: self._waiters += 1 Line 233: self._cond.wait()
To avoid the race explained above:
It seems to me you are right, so I'm doing the change you suggested Line 234: self._waiters -= 1 Line 235: Line 236: def clear(self):
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 22:
addressed Nir's comments.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 22:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10753/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/213... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11695/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/245... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11538/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/227/ : SUCCESS
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 23:
Rebase, no code change.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 23:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10798/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/224... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11740/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/258... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11583/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/239/ : SUCCESS
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 24:
Rebase, no code change.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 24:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10839/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/231... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11781/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/265... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11624/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/246/ : FAILURE
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 26:
rebased. No code changes.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 25:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10886/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/236... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11828/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/270... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11671/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/251/ : SUCCESS
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 26:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/10895/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/238... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/11837/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc19_created/272... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/11680/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/253/ : SUCCESS
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 27:
rebased with no code changes.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 27:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc21_created/129... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/695/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/13856/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/676... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/13067/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/14019/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el7_created/134/ : SUCCESS
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 28:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 28: Verified+1
rebased with no code changes. Copied score.
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 29:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 28:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc21_created/272... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el7_created/278/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/14876/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/837/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/14708/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/13920/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/820... : FAILURE
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 29:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc21_created/274... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el7_created/280/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/14894/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/839/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/14726/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/13938/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/822... : FAILURE
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 30:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 30:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc21_created/287... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el7_created/293/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/14966/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/852/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/14797/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/14009/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/834... : FAILURE
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31: Verified+1
still only rebasing, so copied score.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc21_created/329... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/15205/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/894/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/15036/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/14248/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/877... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el7_created/336/ : SUCCESS
Dan Kenigsberg has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31: Code-Review-1
(7 comments)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 68: self._log.debug('Stopping executor') Line 69: with self._lock: Line 70: self._running = False Line 71: self._tasks.clear() Line 72: for i in range(self._workers_count): i -> _ Line 73: self._tasks.put((_STOP, None)) Line 74: workers = tuple(self._workers) if wait else () Line 75: for worker in workers: Line 76: try:
Line 82: """dispatches a new task to the executor. Line 83: Line 84: The task may be any callable. Line 85: The task will be executed as soon as possible Line 86: in one of the active workers of the executor.""" please document that timeout is measured since callable is called. Line 87: if not self._running: Line 88: raise NotRunning() Line 89: self._tasks.put((callable, timeout)) Line 90:
Line 144: self._scheduler = scheduler Line 145: self._busy = False Line 146: self._discarded = False Line 147: _Worker._id += 1 Line 148: name = "Executor-%i" % _Worker._id if we ever want to have more than one executor, we would need to do a similar trick to enumerate them.
I personally find name = Worker-%i less surprising. Line 149: self._thread = threading.Thread(target=self._run, name=name) Line 150: self._thread.daemon = True Line 151: self._log.debug('Starting worker %s' % name) Line 152: self._thread.start()
Line 154: @property Line 155: def name(self): Line 156: return self._thread.name Line 157: Line 158: def join(self): we attempt to join only the non-busy workers. why? why should we join() at all? Line 159: if self._busy: Line 160: raise _WorkerBusy() Line 161: self._log.debug('Waiting for worker %s', self.name) Line 162: self._thread.join()
Line 155: def name(self): Line 156: return self._thread.name Line 157: Line 158: def join(self): Line 159: if self._busy: race: if join() is called just before _busy was set in _executre_task we can end up joining a busy task. Line 160: raise _WorkerBusy() Line 161: self._log.debug('Waiting for worker %s', self.name) Line 162: self._thread.join() Line 163:
Line 215: # https://docs.python.org/2.6/library/queue.html#Queue.Full Line 216: self._cond = threading.Condition(threading.Lock()) Line 217: self._waiters = 0 Line 218: Line 219: def put(self, task): add a comment
# we don't really mind the race here. in the worse case, we may have # a bit more than _max_tasks. Line 220: if len(self._tasks) == self._max_tasks: Line 221: raise TooManyTasks() Line 222: self._tasks.append(task) Line 223: if self._waiters > 0:
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait() there's a race here if multiple waiting threads have been woken.
can we require pthreading without the spurious wakeups bug, and drop _waiters? otherwise, use a semaphore to count waiters. Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(3 comments)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 144: self._scheduler = scheduler Line 145: self._busy = False Line 146: self._discarded = False Line 147: _Worker._id += 1 Line 148: name = "Executor-%i" % _Worker._id
if we ever want to have more than one executor, we would need to do a simil
Not sure what you sugget, but letting the executor set the name of the workers is probably better. If we have multiple executors each will a name and will name its workers "executor-name-worker-n". And if name is not set we can just use the default thread name. Line 149: self._thread = threading.Thread(target=self._run, name=name) Line 150: self._thread.daemon = True Line 151: self._log.debug('Starting worker %s' % name) Line 152: self._thread.start()
Line 154: @property Line 155: def name(self): Line 156: return self._thread.name Line 157: Line 158: def join(self):
we attempt to join only the non-busy workers. why? why should we join() at
We join when Executor.stop() is invoke with wait == True. This is the default but if you don't care about joining threads you don't have to join them.
If you care about joining we try to join the idle workers, skipping busy workers, so you don't wait forever because a thread is stuck waiting for libvirt. join is racy, and the benefit is questionable, so maybe we should simplify the logic and either join none or all threads.
If we change the logic, the default wait should be False, since waiting on possibly stuck workers is not a good default. Line 159: if self._busy: Line 160: raise _WorkerBusy() Line 161: self._log.debug('Waiting for worker %s', self.name) Line 162: self._thread.join()
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait()
there's a race here if multiple waiting threads have been woken.
The spurious wakeup bug is not related to multiple threads, but to wait returning when notify was not called.
Since a thread acquire self._cond when returning from wait, we don't care if one or more threads wakes up. They will update the _waiters count safely since only one of the threads can continue, holding self._cond.
Can you describe the race? Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
Dan Kenigsberg has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(3 comments)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 144: self._scheduler = scheduler Line 145: self._busy = False Line 146: self._discarded = False Line 147: _Worker._id += 1 Line 148: name = "Executor-%i" % _Worker._id
Not sure what you sugget, but letting the executor set the name of the work
I was suggesting "executor-m-worker-n"; since multiple executors is not a big concern for us right now, I can do with "worker-n". "executor-n", as suggested in the code, is a confusing name for a worker. Line 149: self._thread = threading.Thread(target=self._run, name=name) Line 150: self._thread.daemon = True Line 151: self._log.debug('Starting worker %s' % name) Line 152: self._thread.start()
Line 154: @property Line 155: def name(self): Line 156: return self._thread.name Line 157: Line 158: def join(self):
We join when Executor.stop() is invoke with wait == True. This is the defau
usually, join() is called when you WANT to wait for a certain job to be done. That's pretty useful some times. I cannot find a use case for waiting only for threads that are doing nothing. Line 159: if self._busy: Line 160: raise _WorkerBusy() Line 161: self._log.debug('Waiting for worker %s', self.name) Line 162: self._thread.join()
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait()
The spurious wakeup bug is not related to multiple threads, but to wait ret
I forgot that wait() reacquires the mutex once woken up. There's no race here.
Can you explain why we need self._waiters? What if put() is called right before _waiters += 1? put() would skip the notify, and get() would end up waiting forever. Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait()
I forgot that wait() reacquires the mutex once woken up. There's no race he
Yes, there is a race as you describe if put skip the notify before self._waiters was increased.
self._waiters was introduced to avoid taking the lock and invoking notify when nobody is waiting, which is common when all workers are busy. I profiled this and it was significant improvement when testing with the sampling simulator.
On a busy executor, for example in the content of sampling, this can lead to small delay in one of the tasks, until the next task is scheduled. On an unused executor, this can lead to bigger delay.
I think this can work:
def get(self): while True: try: return self._tasks.popleft() except IndexError: with self._cond: self._waiters += 1 if not self._tasks: self._cond.wait() self._waiters -= 1
If put happened before increasing the self._waiters, self._tasks.empty() will return False, and the waiter will try to get the task from the queue. Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait()
Yes, there is a race as you describe if put skip the notify before self._wa
I should look at the code before replying :-)
There was a race here when the check for empty queue was missing. There is no race with this check. Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(2 comments)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 144: self._scheduler = scheduler Line 145: self._busy = False Line 146: self._discarded = False Line 147: _Worker._id += 1 Line 148: name = "Executor-%i" % _Worker._id
I was suggesting "executor-m-worker-n"; since multiple executors is not a b
worker-n is not good enough since other threads may use this name. I think we should go with simple shcheduler-m-worker-n. Line 149: self._thread = threading.Thread(target=self._run, name=name) Line 150: self._thread.daemon = True Line 151: self._log.debug('Starting worker %s' % name) Line 152: self._thread.start()
Line 154: @property Line 155: def name(self): Line 156: return self._thread.name Line 157: Line 158: def join(self):
usually, join() is called when you WANT to wait for a certain job to be don
So lets remove the racy busy check. Line 159: if self._busy: Line 160: raise _WorkerBusy() Line 161: self._log.debug('Waiting for worker %s', self.name) Line 162: self._thread.join()
Dan Kenigsberg has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait()
I should look at the code before replying :-)
I'm afraid that I've lost track of the argument. I'd appreciate if the _waiters optimization is moved to a follow-up patch where its motivation and racelessness can be discussed separately. Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait()
I'm afraid that I've lost track of the argument. I'd appreciate if the _wai
Francesco, do you want to send a simplified patch and the a waiters optimization on top of it? Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 31:
(2 comments)
http://gerrit.ovirt.org/#/c/29191/31/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 154: @property Line 155: def name(self): Line 156: return self._thread.name Line 157: Line 158: def join(self):
So lets remove the racy busy check.
Will remove. Line 159: if self._busy: Line 160: raise _WorkerBusy() Line 161: self._log.debug('Waiting for worker %s', self.name) Line 162: self._thread.join()
Line 231: except IndexError: Line 232: with self._cond: Line 233: self._waiters += 1 Line 234: if not self._tasks: Line 235: self._cond.wait()
Francesco, do you want to send a simplified patch and the a waiters optimiz
Sure thing. Will change. Line 236: self._waiters -= 1 Line 237: Line 238: def clear(self):
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 32:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
Dan Kenigsberg has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 32: Code-Review+2
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 32:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc21_created/366... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/15430/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/15261/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/14473/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/913... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el7_created/373/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/931/ : SUCCESS
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 32: Code-Review-1
(7 comments)
Could be simplified further and needs cleanup of leftovers.
http://gerrit.ovirt.org/#/c/29191/32/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 81: workers = tuple(self._workers) if wait else () Line 82: for worker in workers: Line 83: try: Line 84: worker.join() Line 85: except _WorkerBusy: Nobody is raising this now - should be deleted. Line 86: self._log.debug('Skipping busy worker %s', worker.name) Line 87: Line 88: def dispatch(self, callable, timeout=None): Line 89: """
Line 138: _STOP = object() Line 139: Line 140: Line 141: class _WorkerBusy(Exception): Line 142: """ Raised when trying to join a busy worker """ Unused, delete Line 143: Line 144: Line 145: class _WorkerDiscarded(Exception): Line 146: """ Raised if worker was discarded during execution of a task """
Line 192: callable() Line 193: except Exception: Line 194: self._log.exception("Unhandled exception in %s", callable) Line 195: finally: Line 196: self._busy = False We can simplify by removing the _busy instance variable, and cancel discard when exiting from the function:
try: callable() except Exception: log it... finally: if discard is not None: discard.cancel() if self._discarded: raise _WorkerDiscarded()
The effect will be the same, worker will be discarded if the discard timer fired before the worker finished and canceled the timer.
We remove unneeded _busy and the check in _discard bellow. Line 197: if self._discarded: Line 198: raise _WorkerDiscarded() Line 199: if discard is not None: Line 200: discard.cancel()
Line 207: def _discard(self): Line 208: if self._discarded: Line 209: raise AssertionError("Attempt to discard worker twice") Line 210: if not self._busy: Line 211: return Do we need this check?
If a worker discard timer fired, the worker should be discarded. I don't see any reason why we check if the worker is busy now. Maybe it was needed when we raise on join? Line 212: self._discarded = True Line 213: self._log.debug("Worker %s discarded", self.name) Line 214: self._executor._worker_discarded(self) Line 215:
Line 214: self._executor._worker_discarded(self) Line 215: Line 216: Line 217: class _TaskQueue(object): Line 218: """ Queue.Queue doesn't support the clear() operation. """ This is not a good docstring. Lets describe what this does and use this line in the body to explain why we do not use Queue.Queue: - It blocks when full - It does not support clear() Line 219: Line 220: def __init__(self, max_tasks): Line 221: self._max_tasks = max_tasks Line 222: self._tasks = collections.deque()
Line 225: # https://docs.python.org/2.6/library/queue.html#Queue.Full Line 226: self._cond = threading.Condition(threading.Lock()) Line 227: Line 228: def put(self, task): Line 229: # there is a race here but we don't really mind. Worst case scenario, there -> There Line 230: # we will have a bit more task then the configured maximum, but we Line 231: # just want to avoid to have indefinite amount of tasks. Line 232: if len(self._tasks) == self._max_tasks: Line 233: raise TooManyTasks()
Line 227: Line 228: def put(self, task): Line 229: # there is a race here but we don't really mind. Worst case scenario, Line 230: # we will have a bit more task then the configured maximum, but we Line 231: # just want to avoid to have indefinite amount of tasks. Please add also that we don't want to block. Otherwise we could use the builtin queue for this. Line 232: if len(self._tasks) == self._max_tasks: Line 233: raise TooManyTasks() Line 234: self._tasks.append(task) Line 235: with self._cond:
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 32:
(7 comments)
Yes, I could have clean it up a bit more. Thanks for the reminder!
http://gerrit.ovirt.org/#/c/29191/32/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 81: workers = tuple(self._workers) if wait else () Line 82: for worker in workers: Line 83: try: Line 84: worker.join() Line 85: except _WorkerBusy:
Nobody is raising this now - should be deleted.
Done Line 86: self._log.debug('Skipping busy worker %s', worker.name) Line 87: Line 88: def dispatch(self, callable, timeout=None): Line 89: """
Line 138: _STOP = object() Line 139: Line 140: Line 141: class _WorkerBusy(Exception): Line 142: """ Raised when trying to join a busy worker """
Unused, delete
Done Line 143: Line 144: Line 145: class _WorkerDiscarded(Exception): Line 146: """ Raised if worker was discarded during execution of a task """
Line 192: callable() Line 193: except Exception: Line 194: self._log.exception("Unhandled exception in %s", callable) Line 195: finally: Line 196: self._busy = False
We can simplify by removing the _busy instance variable, and cancel discard
Done Line 197: if self._discarded: Line 198: raise _WorkerDiscarded() Line 199: if discard is not None: Line 200: discard.cancel()
Line 207: def _discard(self): Line 208: if self._discarded: Line 209: raise AssertionError("Attempt to discard worker twice") Line 210: if not self._busy: Line 211: return
Do we need this check?
Yes, seems now useless. It seems also there is a race between discard timer and the timer disarm in lines 197-200. If so, this was probably added to limit the damage in case of that race. Maybe worth a comment. Line 212: self._discarded = True Line 213: self._log.debug("Worker %s discarded", self.name) Line 214: self._executor._worker_discarded(self) Line 215:
Line 214: self._executor._worker_discarded(self) Line 215: Line 216: Line 217: class _TaskQueue(object): Line 218: """ Queue.Queue doesn't support the clear() operation. """
This is not a good docstring. Lets describe what this does and use this lin
Done Line 219: Line 220: def __init__(self, max_tasks): Line 221: self._max_tasks = max_tasks Line 222: self._tasks = collections.deque()
Line 225: # https://docs.python.org/2.6/library/queue.html#Queue.Full Line 226: self._cond = threading.Condition(threading.Lock()) Line 227: Line 228: def put(self, task): Line 229: # there is a race here but we don't really mind. Worst case scenario,
there -> There
Done Line 230: # we will have a bit more task then the configured maximum, but we Line 231: # just want to avoid to have indefinite amount of tasks. Line 232: if len(self._tasks) == self._max_tasks: Line 233: raise TooManyTasks()
Line 227: Line 228: def put(self, task): Line 229: # there is a race here but we don't really mind. Worst case scenario, Line 230: # we will have a bit more task then the configured maximum, but we Line 231: # just want to avoid to have indefinite amount of tasks.
Please add also that we don't want to block. Otherwise we could use the bui
Done Line 232: if len(self._tasks) == self._max_tasks: Line 233: raise TooManyTasks() Line 234: self._tasks.append(task) Line 235: with self._cond:
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 32:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/32/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 195: finally: Line 196: self._busy = False Line 197: if self._discarded: Line 198: raise _WorkerDiscarded() Line 199: if discard is not None: I don't really like this check. I think it is better and slightly cleaner to always discard timer after succesfull execution. To avoid that we can add an extra-simple private Fake (possibly a singleton to avoid wasteful allocation/deallocation):
class _FakeDiscard(object): def cancel(self): pass
and return this in line 205 below. What do you think? Line 200: discard.cancel() Line 201: Line 202: def _discard_after(self, timeout): Line 203: if timeout is not None:
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 33:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 33:
Address comments from Nir.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 33:
Build Successful
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc21_created/384... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_created/15478/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_pep8_gerrit/15310/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit_tests_gerrit_el/14522/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-fc20_created/931... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el7_created/391/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_install-rpm-sanity-el6_created/949/ : SUCCESS
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 33: Code-Review+1
(1 comment)
http://gerrit.ovirt.org/#/c/29191/33/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 186: finally: Line 187: # there is a race here: even in presence of succesfull Line 188: # execution, discard can fire just before we disarm it. Line 189: # We know and we don't care about sporadic replacements Line 190: # of well-behaving workers. There is no race here - we discard workers that were too slow to disarm the timer. It does not matter if the thread was still blocked on callable when we discard it or it just finished. Line 191: if discard is not None: Line 192: discard.cancel() Line 193: if self._discarded: Line 194: raise _WorkerDiscarded()
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 32:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/32/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 195: finally: Line 196: self._busy = False Line 197: if self._discarded: Line 198: raise _WorkerDiscarded() Line 199: if discard is not None:
I don't really like this check. I think it is better and slightly cleaner t
Isn't this little over-engineered? Why do you care about checking discard?
We can clean this up by moving discard to the object:
self._discard_after(timeout) try: callable() except ... log... finally: self._cancel_discard()
def _start_discard_timeout(self, timeout): self._discard = ...
def _cancel_discard(self): if self._discard is not None: self._discard.cancel() self._discard = None
This can also be easier to debug because we can access the timeout object of the blocked thread (assuming we get into the process using manhole or something similar). Line 200: discard.cancel() Line 201: Line 202: def _discard_after(self, timeout): Line 203: if timeout is not None:
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 33:
(1 comment)
http://gerrit.ovirt.org/#/c/29191/33/lib/vdsm/executor.py File lib/vdsm/executor.py:
Line 186: finally: Line 187: # there is a race here: even in presence of succesfull Line 188: # execution, discard can fire just before we disarm it. Line 189: # We know and we don't care about sporadic replacements Line 190: # of well-behaving workers.
There is no race here - we discard workers that were too slow to disarm the
Right, and your wording make it clear. Will change the comment accordingly. Line 191: if discard is not None: Line 192: discard.cancel() Line 193: if self._discarded: Line 194: raise _WorkerDiscarded()
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 33:
Yes, on second thought my idea is probably leaning toward overengineering. Let's leave this code simple.
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 34:
* Update tracker::IGNORE, no Bug-Url found * Check Bug-Url::WARN, no bug url found, make sure header matches 'Bug-Url: ' and is a valid url. * Check merged to previous::IGNORE, Not in stable branch (['ovirt-3.5', 'ovirt-3.4', 'ovirt-3.3'])
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 34:
this version just improved the wording of a comment, with no code changes.
Nir Soffer has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 34: Code-Review+1
Francesco Romani has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 34: Verified+1
since last verification, changes were only about dropping code. Tests still passing OK, so marking as Verified.
Dan Kenigsberg has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 34: Code-Review+2
Dan Kenigsberg has submitted this change and it was merged.
Change subject: executor: introduce the executor library ......................................................................
executor: introduce the executor library
Executor is a thread pool augmented with the capability of discard blocked worker threads, by replacing them with fresh ones.
This is needed to accomodate the needs of the sampling code. The sampling needs to deal with possibly blocking libvirt calls which needs to enter into the QEMU monitor, thus can get stuck due to QEMU being in D state.
Change-Id: Ic06da1ba57868dc2c7db67a1868ad10087a1cff2 Signed-off-by: Francesco Romani fromani@redhat.com Reviewed-on: http://gerrit.ovirt.org/29191 Reviewed-by: Nir Soffer nsoffer@redhat.com Reviewed-by: Dan Kenigsberg danken@redhat.com --- M debian/vdsm-python.install M lib/vdsm/Makefile.am A lib/vdsm/executor.py M tests/Makefile.am A tests/executorTests.py M vdsm.spec.in 6 files changed, 404 insertions(+), 0 deletions(-)
Approvals: Nir Soffer: Looks good to me, but someone else must approve Dan Kenigsberg: Looks good to me, approved Francesco Romani: Verified
automation@ovirt.org has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 35:
* Update tracker::IGNORE, no Bug-Url found * Set MODIFIED::IGNORE, no Bug-Url found.
oVirt Jenkins CI Server has posted comments on this change.
Change subject: executor: introduce the executor library ......................................................................
Patch Set 35:
Build Failed
http://jenkins.ovirt.org/job/vdsm_master_create-rpms-el7-x86_64_merged/575/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_create-rpms-el6-x86_64_merged/572/ : FAILURE
http://jenkins.ovirt.org/job/vdsm_master_create-rpms-fc20-x86_64_merged/568/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_create-rpms-fc21-x86_64_merged/550/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master-libgfapi_create-rpms-el7-x86_64_mer... : FAILURE
http://jenkins.ovirt.org/job/vdsm_master-libgfapi_create-rpms-el6-x86_64_mer... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master-libgfapi_create-rpms-fc20-x86_64_me... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master-libgfapi_create-rpms-fc21-x86_64_me... : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_unit-tests_merged/4557/ : SUCCESS
http://jenkins.ovirt.org/job/vdsm_master_verify-error-codes_merged/6394/ : FAILURE
vdsm-patches@lists.fedorahosted.org