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

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
+                    # Gracefully terminate this thread if the _stopped
-                    # during unit-test when self._execute() is making db
+                    # attribute was set to true
-                    # access. This makes this thread not terminating and
+                    LOG.info(_("Stopping TaskManager"))
-                    # stop() caller wait indefinitely. So we added a check
-                    # here before trying to do a block call on getting a
-                    # task from queue
                     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:

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,

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):