From 62ec632c47e9d7443789993ee379ee26f3be567d Mon Sep 17 00:00:00 2001
From: Salvatore Orlando <salv.orlando@gmail.com>
Date: Thu, 20 Feb 2014 06:30:11 -0800
Subject: [PATCH] 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 | 27 ++++++++-----------
 neutron/tests/unit/nicira/test_edge_router.py |  3 ++-
 neutron/tests/unit/nicira/test_vcns_driver.py | 12 +++++++++
 3 files changed, 25 insertions(+), 17 deletions(-)

diff --git a/neutron/plugins/nicira/vshield/tasks/tasks.py b/neutron/plugins/nicira/vshield/tasks/tasks.py
index 77f233ae53..4a901f3106 100755
--- 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 ca2d2a33af..2d1adc192c 100644
--- 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 ddc0c339e4..952fdff971 100644
--- 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):