Fixing Slurm/HPC issues (#2222)

* trying to resolve streaming freeze

* preventing Watchdog launch in improper conditions

* docformatter fix

Author: raph-luc

src/ansys/fluent/core/launcher/launcher.py

@@ -9,6 +9,7 @@
 import os
 from pathlib import Path
 import platform
+import socket
 import subprocess
 import tempfile
 import time
@@ -457,8 +458,21 @@ def _generate_launch_string(
     return launch_string
 
 
+def _confirm_watchdog_start(start_watchdog, cleanup_on_exit, fluent_connection):
+    """Confirm whether Fluent is running locally, and whether the Watchdog should be
+    started."""
+    if start_watchdog is None and cleanup_on_exit:
+        host = fluent_connection.connection_properties.cortex_host
+        if host == socket.gethostname():
+            logger.debug(
+                "Fluent running on the host machine and 'cleanup_on_exit' activated, will launch Watchdog."
+            )
+            start_watchdog = True
+    return start_watchdog
+
+
 def scm_to_py(topy, journal_filepaths):
-    """Convert journal filenames to Python filename."""
+    """Convert journal file names to Python file name."""
     fluent_jou_arg = "".join([f'-i "{journal}" ' for journal in journal_filepaths])
     if isinstance(topy, str):
         return f" {fluent_jou_arg} -topy={topy}"
@@ -648,13 +662,6 @@ def launch_fluent(
             "when starting a remote Fluent PyPIM client."
         )
 
-    if (
-        start_watchdog is None
-        and cleanup_on_exit
-        and (fluent_launch_mode in (LaunchMode.CONTAINER, LaunchMode.STANDALONE))
-    ):
-        start_watchdog = True
-
     if dry_run and fluent_launch_mode != LaunchMode.CONTAINER:
         logger.warning(
             "'dry_run' argument for 'launch_fluent' currently is only "
@@ -757,6 +764,9 @@ def launch_fluent(
                 launcher_args=argvals,
                 inside_container=False,
             )
+            start_watchdog = _confirm_watchdog_start(
+                start_watchdog, cleanup_on_exit, session.fluent_connection
+            )
             if start_watchdog:
                 logger.info("Launching Watchdog for local Fluent client...")
                 ip, port, password = _get_server_info(server_info_filepath)
@@ -844,6 +854,8 @@ def launch_fluent(
             )
         )
 
+        if start_watchdog is None and cleanup_on_exit:
+            start_watchdog = True
         if start_watchdog:
             logger.debug("Launching Watchdog for Fluent container...")
             watchdog.launch(os.getpid(), port, password)
@@ -912,11 +924,12 @@ def connect_to_fluent(
     )
     new_session = _get_running_session_mode(fluent_connection)
 
-    if start_watchdog is None and cleanup_on_exit:
-        start_watchdog = True
+    start_watchdog = _confirm_watchdog_start(
+        start_watchdog, cleanup_on_exit, fluent_connection
+    )
 
     if start_watchdog:
-        logger.info("Launching Watchdog for existing Fluent connection...")
+        logger.info("Launching Watchdog for existing Fluent session...")
         ip, port, password = _get_server_info(server_info_filepath, ip, port, password)
         watchdog.launch(os.getpid(), port, password, ip)
 

src/ansys/fluent/core/streaming_services/field_data_streaming.py

@@ -1,6 +1,5 @@
 """Module for Field data streaming."""
 
-import threading
 from typing import Callable, Dict, List, Union
 
 from ansys.api.fluent.v0 import field_data_pb2 as FieldDataProtoModule
@@ -26,7 +25,6 @@ def __init__(self, session_id: str, service):
             streaming_service=service,
         )
         self._session_id: str = session_id
-        self._lock_refresh: threading.Lock = threading.Lock()
 
     def _process_streaming(self, id, stream_begin_method, started_evt, *args, **kwargs):
         """Processes field data streaming."""

src/ansys/fluent/core/streaming_services/streaming.py

@@ -1,7 +1,10 @@
 import itertools
+import logging
 import threading
 from typing import Callable, Optional
 
+logger = logging.getLogger("pyfluent.networking")
+
 
 class StreamingService:
     """Encapsulates a Fluent streaming service."""
@@ -57,7 +60,7 @@ def unregister_callback(self, callback_id: str):
                 del self._service_callbacks[callback_id]
 
     def start(self, *args, **kwargs) -> None:
-        """Start streaming of Fluent transcript."""
+        """Start streaming."""
         with self._lock:
             if not self.is_streaming:
                 self._prepare()
@@ -78,30 +81,14 @@ def start(self, *args, **kwargs) -> None:
                 self._streaming = True
 
     def stop(self) -> None:
-        """Stop streaming of Fluent transcript."""
+        """Stop streaming."""
         if self.is_streaming:
             self._streaming_service.end_streaming(self._id, self._stream_begin_method)
-            self._stream_thread.join()
+            self._stream_thread.join(timeout=5)
+            if self._stream_thread.is_alive():
+                logger.warning(f"Streaming service {self._id} is unresponsive.")
             self._streaming = False
             self._stream_thread = None
 
-    def refresh(self, session_id, event_info) -> None:
-        """Refresh stream.
-
-        Parameters
-        ----------
-        session_id : str
-            Name of the monitor set.
-        event_info : object
-            Event info object.
-
-        Returns
-        -------
-        None
-        """
-        with self._lock_refresh:
-            self.stop()
-            self.start()
-
     def _prepare(self):
         pass  # Currently only used by monitor services.