Fix process monitoring hang on Linux kernels < 5.4 (#273)

Install a SIGCHLD handler via a new _subprocess_install_sigchld_handler C shim so the kernel notifies the monitor thread when children exit. Pass an empty sigmask to epoll_pwait so SIGCHLD is deliverable during the wait, and mark both ends of the self-pipe O_NONBLOCK so the drain loop in _notifyAllKnownChildProcesses exits cleanly on EAGAIN.

Call close_range via a direct syscall wrapper on Linux, with fallback definitions for SYS_close_range and CLOSE_RANGE_CLOEXEC, so the build works against glibc < 2.34 and older kernel-headers (e.g. Amazon Linux 2 with kernel 4.14).

Author: iCharlesHu

Sources/Subprocess/Platforms/Subprocess+Linux.swift

@@ -103,7 +103,9 @@ internal func waitForProcessTermination(
                         }
                         // Now save the registration
                         var newState = storage
-                        newState.continuations[processIdentifier.processDescriptor] = continuation
+                        var list = newState.continuations[processIdentifier.processDescriptor] ?? []
+                        list.append(continuation)
+                        newState.continuations[processIdentifier.processDescriptor] = list
                         state = .started(newState)
                         // No state to resume
                         return nil
@@ -114,10 +116,14 @@ internal func waitForProcessTermination(
                         // is not the case and only save continuation then.
                         switch Result(catching: { () throws(Errno) -> Bool in try processIdentifier.peekIfExited() }) {
                         case let .success(processExited):
-                            // Save this continuation to be called by signal handler
-                            var newState = storage
-                            newState.continuations[processIdentifier.value] = continuation
-                            state = .started(newState)
+                            if !processExited {
+                                // Save this continuation to be called by signal handler
+                                var newState = storage
+                                var list = newState.continuations[processIdentifier.processDescriptor] ?? []
+                                list.append(continuation)
+                                newState.continuations[processIdentifier.value] = list
+                                state = .started(newState)
+                            }
                             return .success(processExited)
                         case let .failure(underlyingError):
                             let error: SubprocessError = .failedToMonitor(
@@ -153,7 +159,7 @@ private enum ProcessMonitorState {
         let epollFileDescriptor: CInt
         let shutdownFileDescriptor: CInt
         let monitorThread: pthread_t
-        var continuations: [PlatformFileDescriptor: CheckedContinuation<Void, any Error>]
+        var continuations: [PlatformFileDescriptor: [CheckedContinuation<Void, any Error>]]
     }
 
     case notStarted
@@ -204,11 +210,7 @@ private func shutdown() {
 /// See the following page for the complete list of `async-signal-safe` functions
 /// https://man7.org/linux/man-pages/man7/signal-safety.7.html
 /// Only these functions can be used in the signal handler below
-private func signalHandler(
-    _ signalNumber: CInt,
-    _ signalInfo: UnsafeMutablePointer<siginfo_t>?,
-    _ context: UnsafeMutableRawPointer?
-) {
+private func signalHandler(_ signalNumber: CInt) {
     let savedErrno = errno
     var one: UInt8 = 1
     withUnsafeBytes(of: &one) { ptr in
@@ -225,7 +227,6 @@ private func monitorThreadFunc(context: MonitorThreadContext) {
     )
     var waitMask = sigset_t()
     sigemptyset(&waitMask)
-    sigaddset(&waitMask, SIGCHLD)
     // Enter the monitor loop
     monitorLoop: while true {
         let eventCount = epoll_pwait(
@@ -247,7 +248,7 @@ private func monitorThreadFunc(context: MonitorThreadContext) {
             let continuations = _processMonitorState.withLock { state -> [CheckedContinuation<Void, any Error>] in
                 let result: [CheckedContinuation<Void, any Error>]
                 if case .started(let storage) = state {
-                    result = Array(storage.continuations.values)
+                    result = storage.continuations.values.flatMap { $0 }
                 } else {
                     result = []
                 }
@@ -333,6 +334,18 @@ private let setup: () = {
         do {
             let (readEnd, writeEnd) = try FileDescriptor.pipe()
             _signalPipe = (readEnd.rawValue, writeEnd.rawValue)
+            // Make the pipe non-blocking. The read end MUST be non-blocking
+            // so the drain loop in _notifyAllKnownChildProcesses exits cleanly
+            // with EAGAIN once the pipe is empty.
+            for fd in [readEnd.rawValue, writeEnd.rawValue] {
+                let existing = fcntl(fd, F_GETFL)
+                if existing < 0 {
+                    throw Errno(rawValue: errno)
+                }
+                if fcntl(fd, F_SETFL, existing | O_NONBLOCK) < 0 {
+                    throw Errno(rawValue: errno)
+                }
+            }
         } catch {
             var underlying: Errno? = nil
             if let err = error as? Errno {
@@ -364,6 +377,11 @@ private let setup: () = {
             _reportFailureWithErrno(errno)
             return
         }
+        // Install the SIGCHLD handler
+        guard _subprocess_install_sigchld_handler(signalHandler) == 0 else {
+            _reportFailureWithErrno(errno)
+            return
+        }
     } else {
         // Mark waitid(P_PIDFD) as supported
         _waitProcessDescriptorSupported = true
@@ -413,9 +431,9 @@ internal func _setupMonitorSignalHandler() {
 
 private func _unregisterProcessDescriptorAndNotify(_ pidfd: CInt, context: MonitorThreadContext) {
     // Remove the continuation
-    let result = _processMonitorState.withLock { state -> (continuation: CheckedContinuation<Void, any Error>, error: SubprocessError?)? in
+    let result = _processMonitorState.withLock { state -> (continuations: [CheckedContinuation<Void, any Error>], error: SubprocessError?)? in
         guard case .started(let storage) = state,
-            let continuation = storage.continuations[pidfd]
+            let continuationList = storage.continuations[pidfd]
         else {
             return nil
         }
@@ -436,23 +454,31 @@ private func _unregisterProcessDescriptorAndNotify(_ pidfd: CInt, context: Monit
             let error = SubprocessError.failedToMonitor(
                 withUnderlyingError: Errno(rawValue: epollErrno)
             )
-            return (continuation, error)
+            return (continuationList, error)
         }
 
-        return (continuation, nil)
+        return (continuationList, nil)
     }
 
-    if let error = result?.error {
-        result?.continuation.resume(throwing: error)
+    guard let result else {
+        return
+    }
+
+    if let error = result.error {
+        for c in result.continuations {
+            c.resume(throwing: error)
+        }
     } else {
-        result?.continuation.resume()
+        for c in result.continuations {
+            c.resume()
+        }
     }
 }
 
 // On older kernels, fall back to using signal handlers
-private typealias ResultContinuation = (
+private typealias ResultContinuations = (
     failure: SubprocessError?,
-    continuation: CheckedContinuation<Void, any Error>
+    continuations: [CheckedContinuation<Void, any Error>]
 )
 private func _notifyAllKnownChildProcesses(_ signalFd: CInt, context: MonitorThreadContext) {
     guard signalFd == _signalPipe.readEnd else {
@@ -467,25 +493,25 @@ private func _notifyAllKnownChildProcesses(_ signalFd: CInt, context: MonitorThr
         }
     }
 
-    let resumingContinuations: [ResultContinuation] = _processMonitorState.withLock { state in
+    let resumingContinuations: [ResultContinuations] = _processMonitorState.withLock { state in
         guard case .started(let storage) = state else {
             return []
         }
         var updatedContinuations = storage.continuations
-        var results: [ResultContinuation] = []
+        var results: [ResultContinuations] = []
         // Since Linux coalesce signals, we need to loop through all known child process
         // to check if they exited.
-        loop: for (knownChildPID, continuation) in storage.continuations {
+        loop: for (knownChildPID, continuations) in storage.continuations {
             switch Result(catching: { () throws(Errno) -> Bool in try _peekIfExited(pid: knownChildPID) }) {
             case let .success(processExisted):
                 if processExisted {
-                    results.append((failure: nil, continuation: continuation))
+                    results.append((failure: nil, continuations: continuations))
                 } else {
                     // The process is still running, move on to the next one
                     continue loop
                 }
             case let .failure(error):
-                results.append((failure: SubprocessError.failedToMonitor(withUnderlyingError: error), continuation: continuation))
+                results.append((failure: SubprocessError.failedToMonitor(withUnderlyingError: error), continuations: continuations))
             }
             // Now we have the exit code, remove saved continuations
             updatedContinuations.removeValue(forKey: knownChildPID)
@@ -497,11 +523,15 @@ private func _notifyAllKnownChildProcesses(_ signalFd: CInt, context: MonitorThr
         return results
     }
     // Resume continuations
-    for c in resumingContinuations {
-        if let error = c.failure {
-            c.continuation.resume(throwing: error)
+    for resumingContinuation in resumingContinuations {
+        if let error = resumingContinuation.failure {
+            for c in resumingContinuation.continuations {
+                c.resume(throwing: error)
+            }
         } else {
-            c.continuation.resume()
+            for c in resumingContinuation.continuations {
+                c.resume()
+            }
         }
     }
 }

Sources/_SubprocessCShims/include/process_shims.h

@@ -120,6 +120,8 @@ int _pidfd_open(pid_t pid);
 #define P_PIDFD 3
 #endif
 
+int _subprocess_install_sigchld_handler(void (* _Nonnull handler)(int));
+
 #endif
 
 #ifdef __cplusplus

Sources/_SubprocessCShims/process_shims.c

@@ -347,6 +347,30 @@ struct linux_dirent64 {
 static int _getdents64(int fd, struct linux_dirent64 *dirp, size_t nbytes) {
     return syscall(SYS_getdents64, fd, dirp, nbytes);
 }
+
+// SYS_close_range is only defined on Linux Kernel 5.9 and above.
+// Define our value if it's not available and call the syscall directly because
+// glibc < 2.34 (e.g. Amazon Linux 2) doesn't provide a close_range() wrapper.
+#ifndef SYS_close_range
+#define SYS_close_range 436
+#endif
+
+#ifndef CLOSE_RANGE_CLOEXEC
+#define CLOSE_RANGE_CLOEXEC (1U << 2)
+#endif
+
+static int _close_range(unsigned int first, unsigned int last, unsigned int flags) {
+    return syscall(SYS_close_range, first, last, flags);
+}
+
+int _subprocess_install_sigchld_handler(void (*handler)(int)) {
+    struct sigaction sa;
+    memset(&sa, 0, sizeof(sa));
+    sa.sa_handler = handler;
+    sa.sa_flags = SA_RESTART | SA_NOCLDSTOP;
+    sigemptyset(&sa.sa_mask);
+    return sigaction(SIGCHLD, &sa, NULL);
+}
 #endif
 
 static pid_t _subprocess_pdfork(int *fdp) {
@@ -674,7 +698,11 @@ int _subprocess_fork_exec(
         // Close all other file descriptors
         rc = -1;
         errno = ENOSYS;
-        #if (__has_include(<linux/close_range.h>) && (!defined(__ANDROID__) || __ANDROID_API__  >= 34)) || defined(__FreeBSD__)
+        #if defined(__linux__)
+        // We must NOT close pipefd[1] for writing errors
+        rc = _close_range(STDERR_FILENO + 1, pipefd[1] - 1, CLOSE_RANGE_CLOEXEC);
+        rc |= _close_range(pipefd[1] + 1, ~0U, CLOSE_RANGE_CLOEXEC);
+        #elif defined(__FreeBSD__)
         // We must NOT close pipefd[1] for writing errors
         rc = close_range(STDERR_FILENO + 1, pipefd[1] - 1, CLOSE_RANGE_CLOEXEC);
         rc |= close_range(pipefd[1] + 1, ~0U, CLOSE_RANGE_CLOEXEC);