client - only initialize shared resources if current_process().name == 'MainProcess'

Author: softwareengineerprogrammer

src/geophires_x_client/__init__.py

@@ -1,9 +1,8 @@
 import atexit
-import os
 import sys
 import threading
 from multiprocessing import Manager
-from pathlib import Path
+from multiprocessing import current_process
 
 # noinspection PyPep8Naming
 from geophires_x import GEOPHIRESv3 as geophires
@@ -36,24 +35,30 @@ def __init__(self, enable_caching=True, logger_name=None):
         self._logger = _get_logger(logger_name=logger_name)
         self._enable_caching = enable_caching
 
-        # Lazy-initialize shared resources if they haven't been already.
         if enable_caching and GeophiresXClient._manager is None:
+            # Lazy-initialize shared resources if they haven't been already.
             self._initialize_shared_resources()
 
     @classmethod
     def _initialize_shared_resources(cls):
         """
         Initializes the multiprocessing Manager and shared resources in a
-        thread-safe manner. It also registers the shutdown hook to ensure
-        automatic cleanup on application exit.
+        thread-safe and now process-safe manner. It also registers the
+        shutdown hook to ensure automatic cleanup on application exit.
         """
-        with cls._init_lock:
-            if cls._manager is None:
-                cls._manager = Manager()
-                cls._cache = cls._manager.dict()
-                cls._lock = cls._manager.RLock()
-                # Register the shutdown method to be called automatically on exit.
-                atexit.register(cls.shutdown)
+        # Ensure that only the top-level user process can create the manager.
+        # A spawned child process, which re-imports this script, will have a different name
+        # (e.g., 'Spawn-1') and will skip this entire block, preventing a recursive crash.
+        if current_process().name == 'MainProcess':
+            with cls._init_lock:
+                if cls._manager is None:
+                    cls._logger = _get_logger(__name__)  # Add a logger for this class method
+                    cls._logger.debug('MainProcess is creating the shared multiprocessing manager...')
+                    cls._manager = Manager()
+                    cls._cache = cls._manager.dict()
+                    cls._lock = cls._manager.RLock()
+                    # Register the shutdown method to be called automatically on exit.
+                    atexit.register(cls.shutdown)
 
     @classmethod
     def shutdown(cls):
@@ -65,9 +70,17 @@ def shutdown(cls):
         """
         with cls._init_lock:
             if cls._manager is not None:
+                cls._logger = _get_logger(__name__)
+                cls._logger.debug('Shutting down the shared multiprocessing manager...')
                 cls._manager.shutdown()
                 # De-register the hook to avoid trying to shut down twice.
-                atexit.unregister(cls.shutdown)
+                try:
+                    atexit.unregister(cls.shutdown)
+                except Exception as e:
+                    # Fails in some environments (e.g. pytest), but is not critical
+                    cls._logger.debug(
+                        f'Encountered exception shutting down the shared multiprocessing manager (OK): ' f'{e!s}'
+                    )
                 cls._manager = None
                 cls._cache = None
                 cls._lock = None
@@ -80,22 +93,23 @@ def get_geophires_result(self, input_params: GeophiresInputParameters) -> Geophi
         """
         is_immutable = isinstance(input_params, ImmutableGeophiresInputParameters)
 
-        if not (self._enable_caching and is_immutable):
+        if not (self._enable_caching and is_immutable and GeophiresXClient._manager is not None):
             return self._run_simulation(input_params)
 
         cache_key = hash(input_params)
 
         with GeophiresXClient._lock:
             if cache_key in GeophiresXClient._cache:
+                # self._logger.debug(f'Cache hit for inputs: {input_params}')
                 return GeophiresXClient._cache[cache_key]
 
+            # Cache miss
             result = self._run_simulation(input_params)
             GeophiresXClient._cache[cache_key] = result
             return result
 
     def _run_simulation(self, input_params: GeophiresInputParameters) -> GeophiresXResult:
         """Helper method to encapsulate the actual GEOPHIRES run."""
-        stash_cwd = Path.cwd()
         stash_sys_argv = sys.argv
         sys.argv = ['', input_params.as_file_path(), input_params.get_output_file_path()]
 
@@ -107,7 +121,6 @@ def _run_simulation(self, input_params: GeophiresInputParameters) -> GeophiresXR
             raise RuntimeError('GEOPHIRES exited without giving a reason') from None
         finally:
             sys.argv = stash_sys_argv
-            os.chdir(stash_cwd)
 
         self._logger.info(f'GEOPHIRES-X output file: {input_params.get_output_file_path()}')
         result = GeophiresXResult(input_params.get_output_file_path())

tests/geophires_x_client_tests/test_imperative_instantiation_in_subprocess.py

@@ -0,0 +1,71 @@
+# ruff: noqa: S603
+
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+from base_test_case import BaseTestCase
+
+
+class GeophiresClientImperativeInstantiationTestCase(BaseTestCase):
+
+    # noinspection PyMethodMayBeStatic
+    def test_imperative_instantiation_in_subprocess(self):
+        """
+        Verifies that GeophiresXClient can be instantiated at the global scope
+        in a script without causing a multiprocessing-related RuntimeError.
+
+        This test directly simulates the failure condition by writing and executing
+        a separate Python script as a subprocess. This ensures that the fix
+        (checking for 'MainProcess') is working correctly on systems that use
+        the 'spawn' start method for multiprocessing (like macOS and Windows).
+        """
+        project_root = Path(__file__).parent.parent.resolve()
+
+        script_content = f"""
+import sys
+# We must add the project root to the path for the import to work.
+sys.path.insert(0, r'{project_root}')
+
+from geophires_x_client import GeophiresXClient
+
+print("Attempting to instantiate GeophiresXClient at the global scope...")
+
+# This is the line that would have previously crashed with a RuntimeError.
+client = GeophiresXClient()
+
+print("Instantiation successful.")
+
+# It is critical to shut down the client to release the manager process,
+# otherwise it can linger and interfere with other tests in the suite.
+GeophiresXClient.shutdown()
+
+print("Shutdown successful.")
+
+# A final message to confirm the script completed without errors.
+print("SUCCESS")
+"""
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            test_script_path = Path(tmpdir) / 'run_client_test.py'
+            test_script_path.write_text(script_content)
+
+            # fmt:off
+            result = subprocess.run(
+                [sys.executable, str(test_script_path)],
+                capture_output=True,
+                text=True,
+                timeout=60
+            )
+            # fmt:on
+
+        assert result.returncode == 0, (
+            f'Subprocess failed with exit code {result.returncode}. This indicates a crash.\\n'
+            f'--- STDOUT ---\\n{result.stdout}\\n'
+            f'--- STDERR ---\\n{result.stderr}'
+        )
+
+        assert 'SUCCESS' in result.stdout, (
+            "Subprocess completed but did not print the final 'SUCCESS' message.\\n" f"--- STDOUT ---\\n{result.stdout}"
+        )