vshield task manager: abort tasks in stop() on termination

This patch kills the manager thread, and aborts active tasks
rather than sending an exception to the manager thread and
have it do the abort on termination.

Unit tests involving vshield task manager might take longer
as a side effect of this patch.

Related-bug: #1282452

Change-Id: I9e9e41ce7e8969a2ea51bfce96b1303125a24308

diff --git a/neutron/plugins/nicira/vshield/tasks/tasks.py b/neutron/plugins/nicira/vshield/tasks/tasks.py
index 77f233ae539634ff16113bd54d6d0c0bb0e79424..4a901f310636b3bff300b153ef2e184b780f7ee8 100755 (executable)
--- a/neutron/plugins/nicira/vshield/tasks/tasks.py
+++ b/neutron/plugins/nicira/vshield/tasks/tasks.py
@@ -20,7 +20,6 @@ import uuid
 
 from eventlet import event
 from eventlet import greenthread
-from eventlet.support import greenlets as greenlet
 
 from neutron.common import exceptions
 from neutron.openstack.common import log as logging
@@ -295,12 +294,9 @@ class TaskManager():
         while True:
             try:
                 if self._stopped:
-                    # Somehow greenlet.GreenletExit exception is ignored
-                    # during unit-test when self._execute() is making db
-                    # access. This makes this thread not terminating and
-                    # stop() caller wait indefinitely. So we added a check
-                    # here before trying to do a block call on getting a
-                    # task from queue
+                    # Gracefully terminate this thread if the _stopped
+                    # attribute was set to true
+                    LOG.info(_("Stopping TaskManager"))
                     break
 
                 # get a task from queue, or timeout for periodic status check
@@ -324,17 +320,11 @@ class TaskManager():
                         self._result(task)
                     else:
                         self._enqueue(task)
-            except greenlet.GreenletExit:
-                break
             except Exception:
-                LOG.exception(_("TaskManager terminated"))
+                LOG.exception(_("TaskManager terminating because "
+                                "of an exception"))
                 break
 
-        self._monitor.stop()
-        if self._monitor_busy:
-            self._monitor.wait()
-        self._abort()
-
     def add(self, task):
         task.id = uuid.uuid1()
         self._tasks_queue.append(task)
@@ -347,8 +337,13 @@ class TaskManager():
             return
         self._stopped = True
         self._thread.kill()
-        self._thread.wait()
         self._thread = None
+        # Stop looping call and abort running tasks
+        self._monitor.stop()
+        if self._monitor_busy:
+            self._monitor.wait()
+        self._abort()
+        LOG.info(_("TaskManager terminated"))
 
     def has_pending_task(self):
         if self._tasks_queue or self._tasks or self._main_thread_exec_task:

diff --git a/neutron/tests/unit/nicira/test_edge_router.py b/neutron/tests/unit/nicira/test_edge_router.py
index ca2d2a33af71d7494e5408a53f85cf0ed42160a0..2d1adc192c324c4615131c688e4911e846279d03 100644 (file)
--- a/neutron/tests/unit/nicira/test_edge_router.py
+++ b/neutron/tests/unit/nicira/test_edge_router.py
@@ -144,7 +144,8 @@ class ServiceRouterTest(test_nicira_plugin.NiciraL3NatTest,
             manager.show_pending_tasks()
             raise Exception(_("Tasks not completed"))
         manager.stop()
-
+        # Ensure the manager thread has been stopped
+        self.assertIsNone(manager._thread)
         super(ServiceRouterTest, self).tearDown()
 
     def _create_router(self, fmt, tenant_id, name=None,

diff --git a/neutron/tests/unit/nicira/test_vcns_driver.py b/neutron/tests/unit/nicira/test_vcns_driver.py
index ddc0c339e4c0a26e235e6235f1f7a00c838019f4..952fdff9718dd1f42a8ae96b4184c8bf6962a841 100644 (file)
--- a/neutron/tests/unit/nicira/test_vcns_driver.py
+++ b/neutron/tests/unit/nicira/test_vcns_driver.py
@@ -45,6 +45,9 @@ class VcnsDriverTaskManagerTestCase(base.BaseTestCase):
 
     def tearDown(self):
         self.manager.stop()
+        # Task manager should not leave running threads around
+        # if _thread is None it means it was killed in stop()
+        self.assertIsNone(self.manager._thread)
         super(VcnsDriverTaskManagerTestCase, self).tearDown()
 
     def _test_task_manager_task_process_state(self, sync_exec=False):
@@ -222,6 +225,9 @@ class VcnsDriverTaskManagerTestCase(base.BaseTestCase):
 
         manager = ts.TaskManager().start(100)
         manager.stop()
+        # Task manager should not leave running threads around
+        # if _thread is None it means it was killed in stop()
+        self.assertIsNone(manager._thread)
         manager.start(100)
 
         alltasks = {}
@@ -236,6 +242,9 @@ class VcnsDriverTaskManagerTestCase(base.BaseTestCase):
 
         greenthread.sleep(stop_wait)
         manager.stop()
+        # Task manager should not leave running threads around
+        # if _thread is None it means it was killed in stop()
+        self.assertIsNone(manager._thread)
 
         for res, tasks in alltasks.iteritems():
             for task in tasks:
@@ -325,6 +334,9 @@ class VcnsDriverTestCase(base.BaseTestCase):
 
     def tearDown(self):
         self.vcns_driver.task_manager.stop()
+        # Task manager should not leave running threads around
+        # if _thread is None it means it was killed in stop()
+        self.assertIsNone(self.vcns_driver.task_manager._thread)
         super(VcnsDriverTestCase, self).tearDown()
 
     def _deploy_edge(self):