Merge pull request #64 from simvue-io/improved_worker

Main process tells worker when to exit

Author: alahiff

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Change log
 
+## v0.10.1
+
+* The worker process now no longer gives a long delay when a run has finished (now at most ~1 second).
+* The worker process ends when the `Run()` context ends or `close` is called, rather than only when the main process exits.
+
 ## v0.10.0
 
 * The `client` class can now be used to retrieve runs.

simvue/__init__.py

@@ -2,4 +2,4 @@
 from simvue.client import Client
 from simvue.handler import Handler
 from simvue.models import RunInput
-__version__ = '0.10.0'
+__version__ = '0.10.1'

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,11 @@ def run(self):
                     pass
                 buffer = []
 
-            if not self._parent_thread.is_alive():
+            if self._shutdown_event.is_set() or not self._parent_thread.is_alive():
                 if self._metrics_queue.empty() and self._events_queue.empty():
                     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() and self._parent_thread.is_alive():
                     time.sleep(1)
                     counter += 1