Change in vdsm[master]: lib: executor: pack items in a Task tuple
fromani at redhat.com
fromani at redhat.com
Fri Nov 6 13:15:49 UTC 2015
Francesco Romani has uploaded a new change for review.
Change subject: lib: executor: pack items in a Task tuple
......................................................................
lib: executor: pack items in a Task tuple
WORK IN PROGRESS
Change-Id: I56dbafd588fc4bde29334e48559255cfcdc0b3ac
Signed-off-by: Francesco Romani <fromani at redhat.com>
---
M lib/vdsm/executor.py
1 file changed, 24 insertions(+), 14 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/92/48192/1
diff --git a/lib/vdsm/executor.py b/lib/vdsm/executor.py
index 767eca3..e852c45 100644
--- a/lib/vdsm/executor.py
+++ b/lib/vdsm/executor.py
@@ -43,6 +43,19 @@
"""Too many tasks for this Executor."""
+WorkItem = collections.namedtuple('WorkItem', ['action', 'timeout', 'name'])
+
+
+def _do_nothing():
+ pass
+
+
+_NULL = WorkItem(_do_nothing, None, 'nothing')
+
+
+_STOP = WorkItem(object(), None, None)
+
+
class Executor(object):
"""
Executes potentially blocking task into background
@@ -80,7 +93,7 @@
self._running = False
self._tasks.clear()
for _ in xrange(self._workers_count):
- self._tasks.put((_STOP, None, None))
+ self._tasks.put(_STOP)
workers = tuple(self._workers) if wait else ()
for worker in workers:
worker.join()
@@ -101,8 +114,8 @@
"""
if not self._running:
raise NotRunning()
- work_id = 'N/A' if workid is None else workid
- self._tasks.put((callable, timeout, work_id))
+ name = 'N/A' if workid is None else workid
+ self._tasks.put(WorkItem(callable, timeout, name))
# Serving workers
@@ -127,10 +140,10 @@
Called from worker thread to get the next task from the taks queue.
Raises NotRunning exception if executor was stopped.
"""
- task, timeout, work_id = self._tasks.get()
+ task = self._tasks.get()
if task is _STOP:
raise NotRunning()
- return task, timeout, work_id
+ return task
# Private
@@ -139,9 +152,6 @@
self._worker_id += 1
worker = _Worker(self, self._scheduler, name)
self._workers.add(worker)
-
-
-_STOP = object()
class _WorkerDiscarded(Exception):
@@ -160,7 +170,7 @@
self._thread.daemon = True
self._log.debug('Starting worker %s' % name)
self._thread.start()
- self._work_id = "N/A"
+ self._task = _NULL
@property
def name(self):
@@ -185,11 +195,10 @@
self._executor._worker_stopped(self)
def _execute_task(self):
- callable, timeout, work_id = self._executor._next_task()
- self._work_id = work_id
- discard = self._discard_after(timeout)
+ self._task = self._executor._next_task()
+ discard = self._discard_after(self._task.timeout)
try:
- callable()
+ self._task.action()
except Exception:
self._log.exception("Unhandled exception in %s", callable)
finally:
@@ -198,6 +207,7 @@
# blocked on callable when we discard it or it just finished.
# However, we expect that most of times only blocked threads
# will be discarded.
+ self._task = _NULL
if discard is not None:
discard.cancel()
if self._discarded:
@@ -213,7 +223,7 @@
raise AssertionError("Attempt to discard worker twice")
self._discarded = True
self._log.debug("Worker %s discarded while doing '%s'",
- self.name, self._work_id)
+ self.name, self._task.name)
self._executor._worker_discarded(self)
--
To view, visit https://gerrit.ovirt.org/48192
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I56dbafd588fc4bde29334e48559255cfcdc0b3ac
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani at redhat.com>
More information about the vdsm-patches
mailing list