Merge pull request #343 from simvue-io/hotfix/fix-join-infinite-loop

Fix infinite thread join bug and resource metrics disabling

Author: kzscisoft

simvue/run.py

@@ -101,6 +101,7 @@ def __init__(self, mode: typing.Literal["online", "offline"] = "online") -> None
         self._simvue: typing.Optional[SimvueBaseClass] = None
         self._pid: typing.Optional[int] = 0
         self._shutdown_event: typing.Optional[threading.Event] = None
+        self._configuration_lock = threading.Lock()
         self._heartbeat_termination_trigger: typing.Optional[threading.Event] = None
         self._storage_id: typing.Optional[str] = None
         self._heartbeat_thread: typing.Optional[threading.Thread] = None
@@ -218,27 +219,27 @@ def _create_heartbeat_callback(
             raise RuntimeError("Could not commence heartbeat, run not initialised")
 
         def _heartbeat(
-            url: typing.Optional[str] = self._url,
-            headers: dict[str, str] = self._headers,
-            run_id: typing.Optional[str] = self._id,
-            online: bool = self._mode == "online",
             heartbeat_trigger: threading.Event = self._heartbeat_termination_trigger,
         ) -> None:
             last_heartbeat = time.time()
             last_res_metric_call = time.time()
 
-            self._add_metrics_to_dispatch(self._get_sysinfo())
-
             while not heartbeat_trigger.is_set():
                 time.sleep(0.1)
 
-                if (
-                    self._resources_metrics_interval
-                    and (res_time := time.time()) - last_res_metric_call
-                    > self._resources_metrics_interval
-                ):
-                    self._add_metrics_to_dispatch(self._get_sysinfo())
-                    last_res_metric_call = res_time
+                with self._configuration_lock:
+                    if (
+                        self._resources_metrics_interval
+                        and (res_time := time.time()) - last_res_metric_call
+                        > self._resources_metrics_interval
+                    ):
+                        # Set join on fail to false as if an error is thrown
+                        # join would be called on this thread and a thread cannot
+                        # join itself!
+                        self._add_metrics_to_dispatch(
+                            self._get_sysinfo(), join_on_fail=False
+                        )
+                        last_res_metric_call = res_time
 
                 if time.time() - last_heartbeat < HEARTBEAT_INTERVAL:
                     continue
@@ -377,20 +378,23 @@ def _start(self, reconnect: bool = False) -> bool:
             self._error(e.args[0])
             return False
 
+        self._active = True
+
         self._dispatcher.start()
         self._heartbeat_thread.start()
 
-        self._active = True
-
         return True
 
-    def _error(self, message: str) -> None:
+    def _error(self, message: str, join_threads: bool = True) -> None:
         """Raise an exception if necessary and log error
 
         Parameters
         ----------
         message : str
             message to display in exception or logger message
+        join_threads : bool
+            whether to join the threads on failure. This option exists to
+            prevent join being called in nested thread calls to this function.
 
         Raises
         ------
@@ -400,7 +404,8 @@ def _error(self, message: str) -> None:
         # Stop heartbeat
         if self._heartbeat_termination_trigger and self._heartbeat_thread:
             self._heartbeat_termination_trigger.set()
-            self._heartbeat_thread.join()
+            if join_threads:
+                self._heartbeat_thread.join()
 
         # Finish stopping all threads
         if self._shutdown_event:
@@ -409,7 +414,8 @@ def _error(self, message: str) -> None:
         # Purge the queue as we can no longer send metrics
         if self._dispatcher and self._dispatcher.is_alive():
             self._dispatcher.purge()
-            self._dispatcher.join()
+            if join_threads:
+                self._dispatcher.join()
 
         if not self._suppress_errors:
             raise RuntimeError(message)
@@ -793,27 +799,28 @@ def config(
             _description_
         """
 
-        if suppress_errors is not None:
-            self._suppress_errors = suppress_errors
+        with self._configuration_lock:
+            if suppress_errors is not None:
+                self._suppress_errors = suppress_errors
 
-        if queue_blocking is not None:
-            self._queue_blocking = queue_blocking
+            if queue_blocking is not None:
+                self._queue_blocking = queue_blocking
 
-        if resources_metrics_interval and disable_resources_metrics:
-            self._error(
-                "Setting of resource metric interval and disabling resource metrics is ambiguous"
-            )
-            return False
+            if resources_metrics_interval and disable_resources_metrics:
+                self._error(
+                    "Setting of resource metric interval and disabling resource metrics is ambiguous"
+                )
+                return False
 
-        if disable_resources_metrics:
-            self._pid = None
-            self._resources_metrics_interval = None
+            if disable_resources_metrics:
+                self._pid = None
+                self._resources_metrics_interval = None
 
-        if resources_metrics_interval:
-            self._resources_metrics_interval = resources_metrics_interval
+            if resources_metrics_interval:
+                self._resources_metrics_interval = resources_metrics_interval
 
-        if storage_id:
-            self._storage_id = storage_id
+            if storage_id:
+                self._storage_id = storage_id
 
         return True
 
@@ -900,6 +907,7 @@ def _add_metrics_to_dispatch(
         step: typing.Optional[int] = None,
         time: typing.Optional[int] = None,
         timestamp: typing.Optional[str] = None,
+        join_on_fail: bool = True,
     ) -> bool:
         if self._mode == "disabled":
             return True
@@ -909,19 +917,21 @@ def _add_metrics_to_dispatch(
             return True
 
         if not self._simvue or not self._dispatcher:
-            self._error("Cannot log metrics, run not initialised")
+            self._error("Cannot log metrics, run not initialised", join_on_fail)
             return False
 
         if not self._active:
-            self._error("Run is not active")
+            self._error("Run is not active", join_on_fail)
             return False
 
         if self._status != "running":
-            self._error("Cannot log metrics when not in the running state")
+            self._error(
+                "Cannot log metrics when not in the running state", join_on_fail
+            )
             return False
 
         if timestamp and not validate_timestamp(timestamp):
-            self._error("Invalid timestamp format")
+            self._error("Invalid timestamp format", join_on_fail)
             return False
 
         _data: dict[str, typing.Any] = {

tests/refactor/test_run_class.py

@@ -38,8 +38,9 @@ def test_log_metrics(
                 name=f"test_run_{str(uuid.uuid4()).split('-', 1)[0]}",
                 tags=["simvue_client_unit_tests"],
                 folder="/simvue_unit_testing",
-                ttl=60 * 60,
-                visibility=visibility
+                retention_period="1 hour",
+                visibility=visibility,
+                resources_metrics_interval=1
             )
         return
 
@@ -48,6 +49,7 @@ def test_log_metrics(
         tags=["simvue_client_unit_tests"],
         folder="/simvue_unit_testing",
         visibility=visibility,
+        resources_metrics_interval=1,
         retention_period="1 hour",
     )
 

tests/refactor/test_scenarios.py

@@ -10,14 +10,18 @@
 @pytest.mark.scenario
 def test_time_multi_run_create_threshold() -> None:
     start = time.time()
-    for i in range(20):
-        with simvue.Run() as run:
-            run.init(
-                f"test run {i}",
-                tags=["test_benchmarking"],
-                folder="/simvue_benchmark_testing",
-                retention_period="1 hour"
-            )
+    runs: list[simvue.Run] = []
+    for i in range(10):
+        run = simvue.Run()
+        run.init(
+            f"test run {i}",
+            tags=["test_benchmarking"],
+            folder="/simvue_benchmark_testing",
+            retention_period="1 hour"
+        )
+        runs.append(run)
+    for run in runs:
+        run.close()
     end = time.time()
     client = simvue.Client()
     with contextlib.suppress(RuntimeError):