Update AsyncProcess with bug fixes

Package.swift

@@ -76,9 +76,22 @@ let package = Package(
       ]
     ),
     .systemLibrary(name: "SystemSQLite", pkgConfig: "sqlite3"),
+
+    // `AsyncProcess` modules and dependencies
+
+    .target(name: "CProcessSpawnSync"),
+    .target(
+      name: "ProcessSpawnSync",
+      dependencies: [
+        "CProcessSpawnSync",
+        .product(name: "Atomics", package: "swift-atomics"),
+        .product(name: "NIOConcurrencyHelpers", package: "swift-nio"),
+      ]
+    ),
     .target(
       name: "AsyncProcess",
       dependencies: [
+        "ProcessSpawnSync",
         .product(name: "Atomics", package: "swift-atomics"),
         .product(name: "AsyncAlgorithms", package: "swift-async-algorithms"),
         .product(name: "Logging", package: "swift-log"),

Sources/AsyncProcess/FileContentStream.swift

@@ -19,7 +19,8 @@ import NIO
 // - Known issues:
 //   - no tests
 //   - most configurations have never run
-struct FileContentStream: AsyncSequence {
+typealias FileContentStream = _FileContentStream
+public struct _FileContentStream: AsyncSequence & Sendable {
   public typealias Element = ByteBuffer
   typealias Underlying = AsyncThrowingChannel<Element, Error>
 
@@ -47,7 +48,17 @@ struct FileContentStream: AsyncSequence {
 
   private let asyncChannel: AsyncThrowingChannel<ByteBuffer, Error>
 
-  public init(
+  public static func makeReader(
+    fileDescriptor: CInt,
+    eventLoop: EventLoop = MultiThreadedEventLoopGroup.singleton.any(),
+    blockingPool: NIOThreadPool = .singleton
+  ) async throws -> _FileContentStream {
+    try await eventLoop.submit {
+      try FileContentStream(fileDescriptor: fileDescriptor, eventLoop: eventLoop, blockingPool: blockingPool)
+    }.get()
+  }
+
+  init(
     fileDescriptor: CInt,
     eventLoop: EventLoop,
     blockingPool: NIOThreadPool? = nil

Sources/AsyncProcess/ProcessExecutor+Convenience.swift

@@ -75,6 +75,7 @@ public extension ProcessExecutor {
     _ arguments: [String],
     standardInput: StandardInput,
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger = ProcessExecutor.disableLogging
   ) async throws -> ProcessExitReason where StandardInput.Element == ByteBuffer {
     let p = Self(
@@ -85,6 +86,7 @@ public extension ProcessExecutor {
       standardInput: standardInput,
       standardOutput: .discard,
       standardError: .discard,
+      teardownSequence: teardownSequence,
       logger: logger
     )
     return try await p.run()
@@ -112,6 +114,7 @@ public extension ProcessExecutor {
     _ arguments: [String],
     standardInput: StandardInput,
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger,
     logConfiguration: OutputLoggingSettings
   ) async throws -> ProcessExitReason where StandardInput.Element == ByteBuffer {
@@ -123,6 +126,7 @@ public extension ProcessExecutor {
       standardInput: standardInput,
       standardOutput: .stream,
       standardError: .stream,
+      teardownSequence: teardownSequence,
       logger: logger
     )
     return try await withThrowingTaskGroup(of: ProcessExitReason?.self) { group in
@@ -179,6 +183,7 @@ public extension ProcessExecutor {
     outputProcessor: @escaping @Sendable (ProcessOutputStream, ByteBuffer) async throws -> (),
     splitOutputIntoLines: Bool = false,
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger = ProcessExecutor.disableLogging
   ) async throws -> ProcessExitReason where StandardInput.Element == ByteBuffer {
     let exe = ProcessExecutor(
@@ -189,6 +194,7 @@ public extension ProcessExecutor {
       standardInput: standardInput,
       standardOutput: .stream,
       standardError: .stream,
+      teardownSequence: teardownSequence,
       logger: logger
     )
     return try await withThrowingTaskGroup(of: ProcessExitReason?.self) { group in
@@ -269,6 +275,7 @@ public extension ProcessExecutor {
     collectStandardError: Bool,
     perStreamCollectionLimitBytes: Int = 128 * 1024,
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger = ProcessExecutor.disableLogging
   ) async throws -> ProcessExitReasonAndOutput where StandardInput.Element == ByteBuffer {
     let exe = ProcessExecutor(
@@ -279,6 +286,7 @@ public extension ProcessExecutor {
       standardInput: standardInput,
       standardOutput: collectStandardOutput ? .stream : .discard,
       standardError: collectStandardError ? .stream : .discard,
+      teardownSequence: teardownSequence,
       logger: logger
     )
 
@@ -351,6 +359,7 @@ public extension ProcessExecutor {
     executable: String,
     _ arguments: [String],
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger = ProcessExecutor.disableLogging
   ) async throws -> ProcessExitReason {
     try await self.run(
@@ -359,6 +368,7 @@ public extension ProcessExecutor {
       arguments,
       standardInput: EOFSequence(),
       environment: environment,
+      teardownSequence: teardownSequence,
       logger: logger
     )
   }
@@ -381,6 +391,7 @@ public extension ProcessExecutor {
     executable: String,
     _ arguments: [String],
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger,
     logConfiguration: OutputLoggingSettings
   ) async throws -> ProcessExitReason {
@@ -390,6 +401,7 @@ public extension ProcessExecutor {
       arguments,
       standardInput: EOFSequence(),
       environment: environment,
+      teardownSequence: teardownSequence,
       logger: logger,
       logConfiguration: logConfiguration
     )
@@ -417,6 +429,7 @@ public extension ProcessExecutor {
     outputProcessor: @escaping @Sendable (ProcessOutputStream, ByteBuffer) async throws -> (),
     splitOutputIntoLines: Bool = false,
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger = ProcessExecutor.disableLogging
   ) async throws -> ProcessExitReason {
     try await self.runProcessingOutput(
@@ -427,6 +440,7 @@ public extension ProcessExecutor {
       outputProcessor: outputProcessor,
       splitOutputIntoLines: splitOutputIntoLines,
       environment: environment,
+      teardownSequence: teardownSequence,
       logger: logger
     )
   }
@@ -455,6 +469,7 @@ public extension ProcessExecutor {
     collectStandardError: Bool,
     perStreamCollectionLimitBytes: Int = 128 * 1024,
     environment: [String: String] = [:],
+    teardownSequence: TeardownSequence = TeardownSequence(),
     logger: Logger = ProcessExecutor.disableLogging
   ) async throws -> ProcessExitReasonAndOutput {
     try await self.runCollectingOutput(
@@ -465,6 +480,7 @@ public extension ProcessExecutor {
       collectStandardError: collectStandardError,
       perStreamCollectionLimitBytes: perStreamCollectionLimitBytes,
       environment: environment,
+      teardownSequence: teardownSequence,
       logger: logger
     )
   }

Sources/AsyncProcess/ProcessExecutor.swift

@@ -10,13 +10,32 @@
 //
 //===----------------------------------------------------------------------===//
 
+import AsyncAlgorithms
 import Atomics
 import Logging
 import NIO
+import ProcessSpawnSync
 
 @_exported import struct SystemPackage.FileDescriptor
 
+#if os(Linux) || ASYNC_PROCESS_FORCE_PS_PROCESS
+// Foundation.Process is too buggy on Linux
+//
+// - Foundation.Process on Linux throws error Error Domain=NSCocoaErrorDomain Code=256 "(null)" if executable not found
+//   https://github.com/swiftlang/swift-corelibs-foundation/issues/4810
+// - Foundation.Process on Linux doesn't correctly detect when child process dies (creating zombie processes)
+//   https://github.com/swiftlang/swift-corelibs-foundation/issues/4795
+// - Foundation.Process on Linux seems to inherit the Process.run()-calling thread's signal mask, even SIGTERM blocked
+//   https://github.com/swiftlang/swift-corelibs-foundation/issues/4772
+typealias Process = PSProcess
+#endif
+
+#if os(iOS) || os(tvOS) || os(watchOS)
+// Note: Process() in iOS/tvOS/watchOS is available in internal builds only under Foundation Private/headers
+import Foundation_Private.NSTask
+#else
 import Foundation
+#endif
 
 public struct ProcessOutputStream: Sendable & Hashable & CustomStringConvertible {
   enum Backing {
@@ -509,12 +528,22 @@ public final actor ProcessExecutor {
     )
 
     p.terminationHandler = { p in
-      let pidExchangeWorked = self.processPid.compareExchange(
-        expected: p.processIdentifier,
-        desired: -1,
-        ordering: .sequentiallyConsistent
-      ).exchanged
-      assert(pidExchangeWorked)
+      let pProcessID = p.processIdentifier
+      var terminationPidExchange: (exchanged: Bool, original: pid_t) = (false, -1)
+      while !terminationPidExchange.exchanged {
+        terminationPidExchange = self.processPid.compareExchange(
+          expected: pProcessID,
+          desired: -1,
+          ordering: .sequentiallyConsistent
+        )
+        if !terminationPidExchange.exchanged {
+          precondition(
+            terminationPidExchange.original == 0,
+            "termination pid exchange failed: \(terminationPidExchange)"
+          )
+          Thread.sleep(forTimeInterval: 0.01)
+        }
+      }
       self.logger.debug(
         "finished running command",
         metadata: [
@@ -560,14 +589,21 @@ public final actor ProcessExecutor {
         ordering: .relaxed
       )
       terminationStreamProducer.finish() // The termination handler will never have fired.
+      try! self.standardOutputWriteHandle?.close()
+      try! self.standardErrorWriteHandle?.close()
       assert(worked) // We just set it to running above, shouldn't be able to race (no `await`).
       assert(original == RunningStateApproximation.running.rawValue) // We compare-and-exchange it.
       throw error
     }
 
     // At this point, the process is running, we should therefore have a process ID (unless we're already dead).
     let childPid = p.processIdentifier
-    _ = self.processPid.compareExchange(expected: 0, desired: childPid, ordering: .sequentiallyConsistent)
+    let runPidExchange = self.processPid.compareExchange(
+      expected: 0,
+      desired: childPid,
+      ordering: .sequentiallyConsistent
+    )
+    precondition(runPidExchange.exchanged, "run pid exchange failed: \(runPidExchange)")
     assert(childPid != 0 || !p.isRunning)
     self.logger.debug(
       "running command",
@@ -658,6 +694,7 @@ public final actor ProcessExecutor {
       }
 
       var exitReason: ProcessExitReason? = nil
+      // cannot fix this warning yet (rdar://113844171)
       while let result = try await runProcessGroup.next() {
         if let result {
           exitReason = result

Sources/CProcessSpawnSync/include/CProcessSpawnSync.h

@@ -0,0 +1,13 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift open source project
+//
+// Copyright (c) 2024 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+#include "ps-api.h"

Sources/CProcessSpawnSync/include/ps-api.h

@@ -0,0 +1,73 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift open source project
+//
+// Copyright (c) 2024 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+#ifndef PS_API_H
+#define PS_API_H
+
+#include <stdbool.h>
+#include <unistd.h>
+
+typedef enum ps_error_kind_s {
+    PS_ERROR_KIND_EXECVE = 1,
+    PS_ERROR_KIND_PIPE = 2,
+    PS_ERROR_KIND_FCNTL = 3,
+    PS_ERROR_KIND_SIGNAL = 4,
+    PS_ERROR_KIND_SIGPROC_MASK = 5,
+    PS_ERROR_KIND_CHDIR = 6,
+    PS_ERROR_KIND_SETSID = 7,
+    PS_ERROR_KIND_DUP2 = 8,
+    PS_ERROR_KIND_READ_FROM_CHILD = 9,
+    PS_ERROR_KIND_DUP = 10,
+    PS_ERROR_KIND_SIGMASK_THREAD = 11,
+} ps_error_kind;
+
+typedef struct ps_error_s {
+    ps_error_kind pse_kind;
+    int pse_code;
+    const char *pse_file;
+    int pse_line;
+    int pse_extra_info;
+} ps_error;
+
+typedef enum ps_fd_setup_kind_s {
+    PS_MAP_FD = 1,
+    PS_CLOSE_FD = 2,
+} ps_fd_setup_kind;
+
+typedef struct ps_fd_setup_s {
+    ps_fd_setup_kind psfd_kind;
+    int psfd_parent_fd;
+} ps_fd_setup;
+
+typedef struct ps_process_configuration_s {
+    const char *psc_path;
+
+    // including argv[0]
+    char **psc_argv;
+
+    char **psc_env;
+
+    const char *psc_cwd;
+
+
+    int psc_fd_setup_count;
+    const ps_fd_setup *psc_fd_setup_instructions;
+
+    bool psc_new_session;
+    bool psc_close_other_fds;
+} ps_process_configuration;
+
+pid_t ps_spawn_process(ps_process_configuration *config, ps_error *out_error);
+
+void ps_convert_exit_status(int in_status, bool *out_has_exited, bool *out_is_exit_code, int *out_code);
+
+#endif