Merge pull request #390 from ExaWorks/proper_singleton_threads

Proper singleton threads

Author: hategan

requirements-tests.txt

@@ -5,4 +5,5 @@
 
 pytest >= 6.2.0
 requests >= 2.25.1
-pytest-cov
+pytest-cov
+pytest-timeout

src/psij/executors/local.py

@@ -7,13 +7,14 @@
 import time
 from abc import ABC, abstractmethod
 from types import FrameType
-from typing import Optional, Dict, List, Type, Tuple
+from typing import Optional, Dict, List, Tuple, Type, cast
 
 import psutil
 
 from psij import InvalidJobException, SubmitException, Launcher
 from psij import Job, JobSpec, JobExecutorConfig, JobState, JobStatus
 from psij import JobExecutor
+from psij.utils import SingletonThread
 
 logger = logging.getLogger(__name__)
 
@@ -22,7 +23,14 @@ def _handle_sigchld(signum: int, frame: Optional[FrameType]) -> None:
     _ProcessReaper.get_instance()._handle_sigchld()
 
 
-_REAPER_SLEEP_TIME = 0.2
+if threading.current_thread() != threading.main_thread():
+    logger.warning('The psij module is being imported from a non-main thread. This prevents the'
+                   'use of signals in the local executor, which will slow things down a bit.')
+else:
+    signal.signal(signal.SIGCHLD, _handle_sigchld)
+
+
+_REAPER_SLEEP_TIME = 0.1
 
 
 class _ProcessEntry(ABC):
@@ -110,18 +118,11 @@ def _get_env(spec: JobSpec) -> Optional[Dict[str, str]]:
         return spec.environment
 
 
-class _ProcessReaper(threading.Thread):
-    _instance: Optional['_ProcessReaper'] = None
-    _lock = threading.RLock()
+class _ProcessReaper(SingletonThread):
 
     @classmethod
     def get_instance(cls: Type['_ProcessReaper']) -> '_ProcessReaper':
-        with cls._lock:
-            if cls._instance is None:
-                cls._instance = _ProcessReaper()
-                cls._instance.start()
-                signal.signal(signal.SIGCHLD, _handle_sigchld)
-            return cls._instance
+        return cast('_ProcessReaper', super().get_instance())
 
     def __init__(self) -> None:
         super().__init__(name='Local Executor Process Reaper', daemon=True)
@@ -198,8 +199,15 @@ class LocalJobExecutor(JobExecutor):
     This job executor is intended to be used when there is no resource manager, only
     the operating system. Or when there is a resource manager, but it should be ignored.
 
-    Limitations: in Linux, attached jobs always appear to complete with a zero exit code regardless
+    Limitations:
+    - In Linux, attached jobs always appear to complete with a zero exit code regardless
     of the actual exit code.
+    - Instantiation of a local executor from both parent process and a `fork()`-ed process
+    is not guaranteed to work. In general, using `fork()` and multi-threading in Linux is unsafe,
+    as suggested by the `fork()` man page. While PSI/J attempts to minimize problems that can
+    arise when `fork()` is combined with threads (which are used by PSI/J), no guarantees can be
+    made and the chances of unexpected behavior are high. Please do not use PSI/J with `fork()`.
+    If you do, please be mindful that support for using PSI/J with `fork()` will be limited.
     """
 
     def __init__(self, url: Optional[str] = None,

src/psij/utils.py

@@ -1,5 +1,7 @@
+import os
+import threading
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Type, Dict
 import sys
 
 
@@ -19,3 +21,32 @@ def path_object_to_full_path(obj: Optional[object]) -> Optional[str]:
             sys.exit("This type " + type(obj).__name__
                      + " for a path is not supported, use pathlib instead")
     return p
+
+
+class SingletonThread(threading.Thread):
+    """
+    A convenience class to return a thread that is guaranteed to be unique to this process.
+
+    This is intended to work with fork() to ensure that each os.getpid() value is associated with
+    at most one thread. This is not safe. The safe thing, as pointed out by the fork() man page,
+    is to not use fork() with threads. However, this is here in an attempt to make it slightly
+    safer for when users really really want to take the risk against all advice.
+    """
+
+    _instances: Dict[int, 'SingletonThread'] = {}
+    _lock = threading.RLock()
+
+    @classmethod
+    def get_instance(cls: Type['SingletonThread']) -> 'SingletonThread':
+        """Returns a started instance of this thread.
+
+        The instance is guaranteed to be unique for this process. This method also guarantees
+        that a forked process will get a separate instance of this thread from the parent.
+        """
+        with cls._lock:
+            my_pid = os.getpid()
+            if my_pid not in cls._instances:
+                instance = cls()
+                cls._instances[my_pid] = instance
+                instance.start()
+            return cls._instances[my_pid]

tests/test_issue_387_1.py

@@ -0,0 +1,11 @@
+import os
+import subprocess
+import sys
+
+import pytest
+
+
+@pytest.mark.timeout(5)
+def test_issue_387_1() -> None:
+    subprocess.run([sys.executable, os.path.abspath(__file__)[:-2] + '.run'],
+                   shell=True, check=True, capture_output=True)

tests/test_issue_387_1.run

@@ -0,0 +1,43 @@
+# This test must be run in a separate process, since it is sensitive
+# to threading and process context.
+
+import logging
+import os
+from multiprocessing import Process, set_start_method
+from threading import Thread
+
+import psij
+
+
+def func():
+    # Get logs from PSI/J
+    logger = logging.getLogger()
+    logger.setLevel("DEBUG")
+    logger.addHandler(logging.StreamHandler())
+
+    exe = psij.JobExecutor.get_instance("local")
+    job = psij.Job(psij.JobSpec("test", "echo", arguments=["foo"]))
+    exe.submit(job)
+    print(job, flush=True)
+
+    # This hangs on the second round
+    job.wait()
+    print(job, flush=True)
+
+
+def fn():
+    print(os.getpid())
+
+
+if __name__ == "__main__":
+    Thread(target=fn).start()
+    set_start_method("fork")
+    p = Process(target=func)
+    p.start()
+    p.join()
+
+    print("Done, and again...")
+
+    p = Process(target=func)
+    p.start()
+    p.join()

tests/test_issue_387_2.py

@@ -0,0 +1,11 @@
+import os
+import subprocess
+import sys
+
+import pytest
+
+
+@pytest.mark.timeout(5)
+def test_issue_387_2() -> None:
+    subprocess.run([sys.executable, os.path.abspath(__file__)[:-2] + '.run'],
+                   shell=True, check=True, capture_output=True)

tests/test_issue_387_2.run

@@ -0,0 +1,26 @@
+# This test must be run in a separate process, since it is sensitive
+# to threading and process context.
+
+import logging
+from threading import Thread
+import psij
+
+
+def func():
+    # Get logs from PSI/J
+    logger = logging.getLogger()
+    logger.setLevel("DEBUG")
+    logger.addHandler(logging.StreamHandler())
+
+    exe = psij.JobExecutor.get_instance("local")
+    job = psij.Job(psij.JobSpec("test", "echo", arguments=["foo"]))
+    exe.submit(job)
+
+    # This hangs
+    job.wait()
+
+if __name__ == "__main__":
+    p = Thread(target=func)
+    p.start()
+    p.join()
+

tests/test_issue_387_3.py

@@ -0,0 +1,11 @@
+import os
+import subprocess
+import sys
+
+import pytest
+
+
+@pytest.mark.timeout(5)
+def test_issue_387_3() -> None:
+    subprocess.run([sys.executable, os.path.abspath(__file__)[:-2] + '.run'],
+                   shell=True, check=True, capture_output=True)

tests/test_issue_387_3.run

@@ -0,0 +1,26 @@
+# This test must be run in a separate process, since it is sensitive
+# to threading and process context.
+
+import logging
+from threading import Thread
+import psij
+
+
+def func():
+    # Get logs from PSI/J
+    logger = logging.getLogger()
+    logger.setLevel("DEBUG")
+    logger.addHandler(logging.StreamHandler())
+
+    exe = psij.JobExecutor.get_instance("local")
+    job = psij.Job(psij.JobSpec("test", "echo", arguments=["foo"]))
+    exe.submit(job)
+
+    # This hangs
+    job.wait()
+
+if __name__ == "__main__":
+    p = Thread(target=func)
+    p.start()
+    p.join()
+