openstack/deb-python-taskflow
Code Issues Proposed changes

Merge "Invert the conductor stop() returned result"

Browse Source
This commit is contained in:
Jenkins
2014-06-01 02:55:56 +00:00
committed by Gerrit Code Review
parent b25e99aa83 059023417e
commit c312ed90bf
2 changed files with 14 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
taskflow/conductors/single_threaded.py
View File

@@ -67,20 +67,22 @@ class SingleThreadedConductor(base.Conductor):
@lock_utils.locked @lock_utils.locked
def stop(self, timeout=None): def stop(self, timeout=None):
"""Stops dispatching and returns whether the dispatcher loop is active """Requests the conductor to stop dispatching and returns whether the
or whether it has ceased. If a timeout is provided the dispatcher stop request was successfully completed. If the dispatching is still
loop may not have ceased by the timeout reached (the request to cease occurring then False is returned otherwise True will be returned to
will be honored in the future). signal that the conductor is no longer dispatching job requests.
NOTE(harlowja): If a timeout is provided the dispatcher loop may
not have ceased by the timeout reached (the request to cease will
be honored in the future) and False will be returned indicating this.
""" """
self._wait_timeout.interrupt() self._wait_timeout.interrupt()
self._dead.wait(timeout) self._dead.wait(timeout)
return self.dispatching return self._dead.is_set()
@property @property
def dispatching(self): def dispatching(self):
if self._dead.is_set(): return not self._dead.is_set()
return False
return True
def _dispatch_job(self, job): def _dispatch_job(self, job):
engine = self._engine_from_job(job) engine = self._engine_from_job(job)

7
taskflow/tests/unit/conductor/test_conductor.py
View File

@@ -88,7 +88,8 @@ class SingleThreadedConductorTest(test_utils.EngineTestBase, test.TestCase):
with close_many(components.conductor, components.client): with close_many(components.conductor, components.client):
t = make_thread(components.conductor) t = make_thread(components.conductor)
t.start() t.start()
self.assertFalse(components.conductor.stop(0.5)) self.assertTrue(components.conductor.stop(0.5))
self.assertFalse(components.conductor.dispatching)
t.join() t.join()
def test_run(self): def test_run(self):
@@ -111,7 +112,7 @@ class SingleThreadedConductorTest(test_utils.EngineTestBase, test.TestCase):
details={'flow_uuid': fd.uuid}) details={'flow_uuid': fd.uuid})
consumed_event.wait(1.0) consumed_event.wait(1.0)
self.assertTrue(consumed_event.is_set()) self.assertTrue(consumed_event.is_set())
components.conductor.stop(1.0) self.assertTrue(components.conductor.stop(1.0))
self.assertFalse(components.conductor.dispatching) self.assertFalse(components.conductor.dispatching)
persistence = components.persistence persistence = components.persistence
@@ -142,7 +143,7 @@ class SingleThreadedConductorTest(test_utils.EngineTestBase, test.TestCase):
details={'flow_uuid': fd.uuid}) details={'flow_uuid': fd.uuid})
consumed_event.wait(1.0) consumed_event.wait(1.0)
self.assertTrue(consumed_event.is_set()) self.assertTrue(consumed_event.is_set())
components.conductor.stop(1.0) self.assertTrue(components.conductor.stop(1.0))
self.assertFalse(components.conductor.dispatching) self.assertFalse(components.conductor.dispatching)
persistence = components.persistence persistence = components.persistence