HeatStack async mode fix

Improve asynchronous push mode of HeatStack:
- Use spawn_after instead of spawn_after_local. Otherwise the data is never
   pushed if the initiated thread were to exit
- Cancel background thread instead of killing it. Cancel cancels the thread
   only if it hasn't started yet instead of killing it somewhere in the middle.
- Add post-execution cleanup to guarantee that async data push happens
   before the execution session end
- Make Instance destruction use async push to speed up the destruction
   in case when there are many servers and to test the HeatStack async mode

Closes-Bug: #1643702
Change-Id: I11d157844cb1d973d2cac62c2e6d67d047f75164

meta/io.murano/Classes/resources/Instance.yaml

@@ -332,7 +332,7 @@ Methods:
       - $template: $region.stack.current()
       - If: $template.get(resources) and $template.get(outputs) and not sys:GC.isDoomed($region)
         Then:
-          - $region.stack.push()
+          - $region.stack.push(async => true)
       - $.setAttr(instanceResources, [])
       - $.setAttr(instanceOutputs, [])
       - $.setAttr(fipAssigned, false)

murano/engine/system/heat_stack.py

@@ -59,7 +59,8 @@ class HeatStack(object):
 
     def _kill_push_thread(self):
         if self._is_push_thread_alive():
-            self._push_thread.kill()
+            self._push_thread.cancel()
+            self._wait_push_thread()
 
     def _wait_push_thread(self):
         if not self._is_push_thread_alive():
@@ -271,8 +272,18 @@ class HeatStack(object):
 
         self._kill_push_thread()
         if async:
+            if self._push_thread is None:
+
+                def cleanup():
+                    try:
+                        self._wait_push_thread()
+                    finally:
+                        self._push_thread = None
+
+                dsl.get_execution_session().on_session_finish(cleanup)
+
             self._push_thread =\
-                eventlet.greenthread.spawn_after_local(
+                eventlet.greenthread.spawn_after(
                     1, self._push, helpers.get_object_store())
         else:
             self._push()

murano/tests/unit/test_heat_stack.py

@@ -198,7 +198,8 @@ class TestHeatStack(base.MuranoTestCase):
         hs._hot_environment = 'environments'
         hs._parameters = {}
         hs._applied = False
-        hs.push(async=True)
+        with mock.patch('murano.dsl.dsl.get_execution_session'):
+            hs.push(async=True)
 
         expected_template = {
             'heat_template_version': '2013-05-23',