Use a shared cache across instances of GeophiresXClient so consumers don't have to manage singletons

Author: softwareengineerprogrammer

src/geophires_x_client/__init__.py

@@ -1,7 +1,7 @@
 import json
 import os
 import sys
-import threading
+from multiprocessing import Manager
 from pathlib import Path
 
 # noinspection PyPep8Naming
@@ -14,51 +14,57 @@
 
 
 class GeophiresXClient:
+    # Use a multiprocessing Manager to create a cache and lock that are
+    # shared across all processes spawned by a ProcessPoolExecutor.
+    _manager = Manager()
+    _cache = _manager.dict()
+    _lock = _manager.Lock()
+
     def __init__(self, enable_caching=True, logger_name=None):
         if logger_name is None:
             logger_name = __name__
 
         self._logger = _get_logger(logger_name=logger_name)
-        self._enable_caching = enable_caching
-        self._cache = {}
-        self._lock = threading.Lock()
+        self.enable_caching = enable_caching
 
     def get_geophires_result(self, input_params: GeophiresInputParameters) -> GeophiresXResult:
-        """
-        Calculates a GEOPHIRES result in a thread-safe manner.
-
-        This method ensures thread safety by using a lock to serialize access,
-        preventing race conditions on the cache and corruption of global state
-        (sys.argv, os.cwd) that GEOPHIRES modifies.
-        """
-        with self._lock:
-            cache_key = hash(input_params)
-            if self._enable_caching and cache_key in self._cache:
-                return self._cache[cache_key]
-
-            stash_cwd = Path.cwd()
-            stash_sys_argv = sys.argv
-
-            sys.argv = ['', input_params.as_file_path(), input_params.get_output_file_path()]
-            try:
-                geophires.main(enable_geophires_logging_config=False)
-            except Exception as e:
-                raise RuntimeError(f'GEOPHIRES encountered an exception: {e!s}') from e
-            except SystemExit:
-                raise RuntimeError('GEOPHIRES exited without giving a reason') from None
-            finally:
-                # Ensure global state is restored even if geophires.main() fails
-                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())
-            if self._enable_caching:
-                self._cache[cache_key] = result
-
+        if not self.enable_caching:
+            return self._run_geophires(input_params)
+
+        cache_key = hash(input_params)
+        with self.__class__._lock:
+            # Use a lock to ensure the check-and-set operation is atomic.
+            if cache_key in self.__class__._cache:
+                # Cache hit
+                return self.__class__._cache[cache_key]
+
+            # If not in cache, we will run the simulation, still under the lock,
+            # to prevent a race condition with other processes.
+            # Cache miss
+            result = self._run_geophires(input_params)
+            self.__class__._cache[cache_key] = result
             return result
 
+    def _run_geophires(self, input_params: GeophiresInputParameters) -> GeophiresXResult:
+        """Helper method to encapsulate the actual GEOPHIRES execution."""
+        stash_cwd = Path.cwd()
+        stash_sys_argv = sys.argv
+
+        sys.argv = ['', input_params.as_file_path(), input_params.get_output_file_path()]
+        try:
+            geophires.main(enable_geophires_logging_config=False)
+        except Exception as e:
+            raise RuntimeError(f'GEOPHIRES encountered an exception: {e!s}') from e
+        except SystemExit:
+            raise RuntimeError('GEOPHIRES exited without giving a reason') from None
+        finally:
+            # Ensure we always restore the original state
+            sys.argv = stash_sys_argv
+            os.chdir(stash_cwd)
+
+        self._logger.info(f'GEOPHIRES-X output file: {input_params.get_output_file_path()}')
+        return GeophiresXResult(input_params.get_output_file_path())
+
 
 if __name__ == '__main__':
     client = GeophiresXClient()

tests/geophires_x_client_tests/test_geophires_client_caching.py

@@ -56,9 +56,12 @@ def test_caching_with_identical_immutable_params(self, mock_geophires_main: unit
         # The core assertion: was the expensive simulation function only called once?
         mock_geophires_main.assert_called_once()
 
-        # The results should not only be equivalent but should be the *same object*
-        # retrieved from the cache on the second call.
-        self.assertIs(result1, result2, 'The second result should be the cached object instance.')
+        self.assertDictEqual(result1.result, result2.result)
+
+        # TODO The results should probably not only be equivalent but also the *same object*...
+        #  For now they not, but we probably don't care about this since the important part is performance/cache hit -
+        #  manually verified the cache hit in debugger during development.
+        # self.assertIs(result1, result2, 'The second result should be the cached object instance.')
 
     @patch('geophires_x_client.geophires.main')
     def test_no_caching_with_different_immutable_params(self, mock_geophires_main: unittest.mock.MagicMock):