Clean up signalling events after stack updates

We provide an Event object to every stack update in order to signal it that
we need to cancel the update; this change ensures that we delete it when
the thread is complete. Previously we were leaking memory on every update.

Change-Id: Ie0fbe3ec3ab57a2cb9dd5e551db15285b2b423c0
Closes-Bug: #1376857

heat/engine/service.py

@@ -118,7 +118,9 @@ def start_with_lock(self, cnxt, stack, engine_id, func, *args, **kwargs):
         """
         lock = stack_lock.StackLock(cnxt, stack, engine_id)
         with lock.thread_lock(stack.id):
-            self.start_with_acquired_lock(stack, lock, func, *args, **kwargs)
+            th = self.start_with_acquired_lock(stack, lock,
+                                               func, *args, **kwargs)
+            return th
 
     def start_with_acquired_lock(self, stack, lock, func, *args, **kwargs):
         """
@@ -143,6 +145,7 @@ def release(gt, *args):
 
         th = self.start(stack.id, func, *args, **kwargs)
         th.link(release, stack.id)
+        return th
 
     def add_timer(self, stack_id, func, *args, **kwargs):
         """
@@ -157,6 +160,11 @@ def add_timer(self, stack_id, func, *args, **kwargs):
     def add_event(self, stack_id, event):
         self.events[stack_id].append(event)
 
+    def remove_event(self, stack_id, event):
+        for e in self.events.pop(stack_id, []):
+            if e is not event:
+                self.add_event(stack_id, e)
+
     def stop_timers(self, stack_id):
         if stack_id in self.groups:
             self.groups[stack_id].stop_timers()
@@ -687,12 +695,13 @@ def update_stack(self, cnxt, stack_identity, template, params,
         updated_stack.validate()
 
         event = eventlet.event.Event()
+        th = self.thread_group_mgr.start_with_lock(cnxt, current_stack,
+                                                   self.engine_id,
+                                                   current_stack.update,
+                                                   updated_stack,
+                                                   event=event)
+        th.link(self.thread_group_mgr.remove_event, current_stack.id, event)
         self.thread_group_mgr.add_event(current_stack.id, event)
-        self.thread_group_mgr.start_with_lock(cnxt, current_stack,
-                                              self.engine_id,
-                                              current_stack.update,
-                                              updated_stack,
-                                              event=event)
         return dict(current_stack.identifier())
 
     @request_context

heat/tests/test_engine_service.py

@@ -1792,6 +1792,8 @@ def test_stack_event_list_deleted_resource(self):
 
         thread = self.m.CreateMockAnything()
         thread.link(mox.IgnoreArg(), self.stack.id).AndReturn(None)
+        thread.link(mox.IgnoreArg(), self.stack.id,
+                    mox.IgnoreArg()).AndReturn(None)
 
         def run(stack_id, func, *args, **kwargs):
             func(*args)
@@ -3482,6 +3484,17 @@ def test_tgm_add_event(self):
         thm.add_event(stack_id, e2)
         self.assertEqual(thm.events[stack_id], [e1, e2])
 
+    def test_tgm_remove_event(self):
+        stack_id = 'add_events_test'
+        e1, e2 = mock.Mock(), mock.Mock()
+        thm = service.ThreadGroupManager()
+        thm.add_event(stack_id, e1)
+        thm.add_event(stack_id, e2)
+        thm.remove_event(stack_id, e2)
+        self.assertEqual(thm.events[stack_id], [e1])
+        thm.remove_event(stack_id, e1)
+        self.assertNotIn(stack_id, thm.events)
+
     def test_tgm_send(self):
         stack_id = 'send_test'
         e1, e2 = mock.MagicMock(), mock.Mock()