openstack/deb-python-taskflow
Code Issues Proposed changes

Disallowing starting the executor when worker running

Browse Source 
To avoid consistency/threading/runtime issues stop the action
engine executor from being started if it is already running with
a valid worker thread.

Change-Id: I39925e55e7b171f289152d941ebdf390552f880c
This commit is contained in:
Joshua Harlow
2014-12-30 10:29:55 -08:00
committed by Joshua Harlow
parent d6a1be73e8
commit 2ed1ad9490
1 changed files with 8 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
taskflow/engines/action_engine/executor.py
View File

@@ -444,6 +444,9 @@ class ParallelProcessTaskExecutor(ParallelTaskExecutor):
return futures.ProcessPoolExecutor(max_workers=max_workers)
def start(self):
if threading_utils.is_alive(self._worker):
raise RuntimeError("Worker thread must be stopped via stop()"
" before starting/restarting")
super(ParallelProcessTaskExecutor, self).start()
# TODO(harlowja): do something else here besides accessing a state
# of the manager internals (it doesn't seem to expose any way to know
@@ -452,12 +455,11 @@ class ParallelProcessTaskExecutor(ParallelTaskExecutor):
self._manager = multiprocessing.Manager()
if self._manager._state.value == managers.State.INITIAL:
self._manager.start()
if not threading_utils.is_alive(self._worker):
self._dispatcher.reset()
self._queue = self._manager.Queue()
self._worker = threading_utils.daemon_thread(self._dispatcher.run,
self._queue)
self._worker.start()
self._dispatcher.reset()
self._queue = self._manager.Queue()
self._worker = threading_utils.daemon_thread(self._dispatcher.run,
self._queue)
self._worker.start()
def stop(self):
self._dispatcher.interrupt()