Use multiprocessing.Event to tell worker when to exit

Author: alahiff

simvue/run.py

@@ -162,6 +162,7 @@ def __enter__(self):
 
     def __exit__(self, type, value, traceback):
         if self._name and self._status == 'running':
+            self._shutdown_event.set()
             if not type:
                 self.set_status('completed')
             else:
@@ -203,8 +204,10 @@ def _start(self, reconnect=False):
 
         self._metrics_queue = multiprocessing.Manager().Queue(maxsize=self._queue_size)
         self._events_queue = multiprocessing.Manager().Queue(maxsize=self._queue_size)
+        self._shutdown_event = multiprocessing.Manager().Event()
         self._worker = Worker(self._metrics_queue,
                               self._events_queue,
+                              self._shutdown_event,
                               self._uuid,
                               self._name,
                               self._url,
@@ -659,8 +662,9 @@ def close(self):
         if not self._active:
             self._error('Run is not active')
             return False
-
+   
         self.set_status('completed')
+        self._shutdown_event.set()
 
     def set_folder_details(self, path, metadata={}, tags=[], description=None):
         """

simvue/worker.py

@@ -29,11 +29,12 @@ def update_processes(parent, processes):
 
 
 class Worker(threading.Thread):
-    def __init__(self, metrics_queue, events_queue, uuid, run_name, url, headers, mode, pid, resources_metrics_interval):
+    def __init__(self, metrics_queue, events_queue, shutdown_event, uuid, run_name, url, headers, mode, pid, resources_metrics_interval):
         threading.Thread.__init__(self)
         self._parent_thread = threading.currentThread()
         self._metrics_queue = metrics_queue
         self._events_queue = events_queue
+        self._shutdown_event = shutdown_event
         self._run_name = run_name
         self._uuid = uuid
         self._url = url
@@ -144,11 +145,12 @@ def run(self):
                     pass
                 buffer = []
 
-            if not self._parent_thread.is_alive():
+            if self._shutdown_event.is_set():
                 if self._metrics_queue.empty() and self._events_queue.empty():
+                    print('eixitig')
                     sys.exit(0)
             else:
                 counter = 0
-                while counter < POLLING_INTERVAL and self._parent_thread.is_alive():
+                while counter < POLLING_INTERVAL and not self._shutdown_event.is_set():
                     time.sleep(1)
                     counter += 1