Use per-process singleton threads for the local executor process reaper.

The issue is that singletons based on flags do not work properly with
fork() since memory is copied to the child process. The only reasonable
way to distinguish between the parent and child is to look at the PID.
That said, using `fork()` with multi-threading is a bad idea. This commit
is not an endorsement but an attempt to minimize damage when possible.

Author: hategan

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__)
 
@@ -117,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)
@@ -205,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]