Attmept to use cross process critical section

simonrw · simonrw · commit 7ee127e5c77c · 2025-05-15T17:25:09.000+01:00
diff --git a/localstack-core/localstack/testing/pytest/in_memory_localstack.py b/localstack-core/localstack/testing/pytest/in_memory_localstack.py
@@ -35,9 +35,6 @@ def pytest_configure(config):
 _started = threading.Event()
 
 
-critical_section = CrossProcessCriticalSection()
-
-
 def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager):
     parser.addoption(
         "--start-localstack",
@@ -79,14 +76,25 @@ def pytest_runtestloop(session: Session):
 
     from localstack.runtime import current
 
-    _started.set()
-    runtime = current.initialize_runtime()
-    # start runtime asynchronously
-    threading.Thread(target=runtime.run).start()
+    def work():
+        _started.set()
+        runtime = current.initialize_runtime()
+
+        # start runtime asynchronously
+        def target():
+            try:
+                runtime.run()
+            except Exception:
+                LOG.warning("error starting runtime", exc_info=LOG.isEnabledFor(logging.DEBUG))
+
+        threading.Thread(target=target).start()
 
-    # wait for runtime to be ready
-    if not runtime.ready.wait(timeout=120):
-        raise TimeoutError("gave up waiting for runtime to be ready")
+        # wait for runtime to be ready
+        if not runtime.ready.wait(timeout=120):
+            raise TimeoutError("gave up waiting for runtime to be ready")
+
+    critical_section = CrossProcessCriticalSection("runtime.start", work)
+    critical_section.run_once()
 
 
 @pytest.hookimpl(trylast=True)
@@ -103,9 +111,16 @@ def pytest_sessionfinish(session: Session):
         LOG.warning("Could not access the current runtime in a pytest sessionfinish hook.")
         return
 
-    get_current_runtime().shutdown()
-    LOG.info("waiting for runtime to stop")
+    def work():
+        get_current_runtime().shutdown()
+        LOG.info("waiting for runtime to stop")
+
+        # wait for runtime to shut down
+        if not get_current_runtime().stopped.wait(timeout=20):
+            LOG.warning("gave up waiting for runtime to stop, returning anyway")
 
-    # wait for runtime to shut down
-    if not get_current_runtime().stopped.wait(timeout=20):
-        LOG.warning("gave up waiting for runtime to stop, returning anyway")
+    critical_section = CrossProcessCriticalSection("runtime.shutdown", work)
+    try:
+        critical_section.run_once()
+    except Exception:
+        LOG.warning("error shutting runtime down", exc_info=LOG.isEnabledFor(logging.DEBUG))
diff --git a/localstack-core/localstack/utils/processlock.py b/localstack-core/localstack/utils/processlock.py
@@ -1,51 +1,73 @@
 import atexit
+import logging
 import os
+import time
 from pathlib import Path
 from typing import Callable
 
 from filelock import FileLock
 
-LOCKFILE = Path("/tmp/my_task.lock")
-DONEFILE = Path("/tmp/my_task.done")
-PIDFILE = Path("/tmp/my_task.pids")
-LOCK_TIMEOUT = 60  # seconds
+from localstack.logging.setup import setup_logging_from_config
+from localstack.utils.functions import run_safe
+
+LOG = logging.getLogger(__name__)
+logging.getLogger("filelock").setLevel(logging.WARNING)
 
 
 class CrossProcessCriticalSection:
-    def __init__(self, work: Callable[[], None] | None = None):
+    def __init__(self, name: str, work: Callable[[], None]):
+        self.name = name
         self.work = work
-        self.pid = os.getpid()
+        self.lockfile = Path(f"/tmp/my_task.{name}.lock")
+        self.donefile = Path(f"/tmp/my_task.{name}.done")
+        self.lock_timeout = 60  # seconds
 
-    def _log(self, message: str):
-        print(f"[{self.pid}]: {message}")
+        self.pid = os.getpid()
 
     def run_once(self):
-        if not self.work:
-            raise RuntimeError("Work not defined")
-
-        self._log("waiting for file lock")
-        with FileLock(str(LOCKFILE), timeout=LOCK_TIMEOUT):
-            self._log("got file lock")
-            if DONEFILE.is_file():
-                self._log("donefile exists, exiting critical section")
+        self._log("waiting for lock")
+        with FileLock(str(self.lockfile), timeout=self.lock_timeout):
+            self._log("lock achieved")
+            if self.donefile.is_file():
+                self._log("work already complete")
                 # we are not the first process so end this critical section
                 return
 
             # we are the first to reach this critical section, so make sure the done file does not exist
             # and run the work
-            self._log("donefile does not exist, registering cleanup")
+            self._log("registering atexit")
             atexit.register(self.cleanup)
 
-            self._log("adding donefile")
-            with DONEFILE.open("w") as outfile:
+            with self.donefile.open("w") as outfile:
                 outfile.write("1")
 
-            self._log("performing work")
+            self._log("starting work")
             self.work()
             self._log("work finished")
 
-        self._log("ended critical section")
+            run_safe(lambda: os.remove(self.lockfile))
 
     def cleanup(self):
         self._log("cleaning up")
-        os.remove(DONEFILE)
+        for path in [self.lockfile, self.donefile]:
+            run_safe(lambda: os.remove(path))
+
+    def _log(self, message: str):
+        LOG.debug("[%s:%d] %s", self.name, self.pid, message)
+
+
+def __main__():
+    setup_logging_from_config()
+
+    def work():
+        print("Should only run once")
+        time.sleep(10)
+
+    cpcs = CrossProcessCriticalSection(work)
+    cpcs.run_once()
+
+    time.sleep(5)
+
+
+if __name__ == "__main__":
+    __main__()
