Fixed hang when cat long files

Author: iCharlesHu

Sources/FoundationEssentials/Subprocess/Platforms/Subprocess+Unix.swift

@@ -26,6 +26,8 @@ import SystemPackage
 package import _CShims
 #endif
 
+import Dispatch
+
 
 // MARK: - Signals
 extension Subprocess {
@@ -270,17 +272,40 @@ extension Subprocess.Configuration {
 @Sendable
 internal func monitorProcessTermination(
     forProcessWithIdentifier pid: Subprocess.ProcessIdentifier
-) -> Subprocess.TerminationStatus {
-    var status: Int32 = -1
-    // Block and wait
-    waitpid(pid.value, &status, 0)
-    if _was_process_exited(status) != 0 {
-        return .exit(_get_exit_code(status))
+) async -> Subprocess.TerminationStatus {
+    return await withCheckedContinuation { continuation in
+        let source = DispatchSource.makeProcessSource(
+            identifier: pid.value,
+            eventMask: [.exit, .signal]
+        )
+        source.setEventHandler {
+            source.cancel()
+            var status: Int32 = -1
+            waitpid(pid.value, &status, WNOHANG)
+            if _was_process_exited(status) != 0 {
+                continuation.resume(returning: .exit(_get_exit_code(status)))
+                return
+            }
+            if _was_process_signaled(status) != 0 {
+                continuation.resume(returning: .unhandledException(_get_signal_code(status)))
+                return
+            }
+            fatalError("Unexpected exit status type: \(status)")
+        }
+        source.resume()
     }
-    if _was_process_signaled(status) != 0 {
-        return .unhandledException(_get_signal_code(status))
+}
+
+// MARK: - Read Buffer Size
+extension Subprocess {
+    @inline(__always)
+    internal static var readBufferSize: Int {
+#if canImport(Darwin)
+        return 16384
+#else
+        return Platform.pageSize
+#endif // canImport(Darwin)
     }
-    fatalError("Unexpected exit status type: \(status)")
 }
 
 #endif // canImport(Darwin) || canImport(Glibc)

Sources/FoundationEssentials/Subprocess/Subprocess+AsyncBytes.swift

@@ -14,9 +14,6 @@ import Dispatch
 
 extension Subprocess {
     public struct AsyncBytes: AsyncSequence, Sendable {
-        @inline(__always) static var bufferSize: Int {
-            16384
-        }
         public typealias Element = UInt8
 
         @_nonSendable
@@ -42,9 +39,9 @@ extension Subprocess {
                 try Task.checkCancellation()
                 do {
                     self.buffer = try await self.fileDescriptor.read(
-                        upToLength: AsyncBytes.bufferSize)
+                        upToLength: Subprocess.readBufferSize)
                     self.currentPosition = 0
-                    if self.buffer.count < AsyncBytes.bufferSize {
+                    if self.buffer.count < Subprocess.readBufferSize {
                         self.finished = true
                     }
                 } catch {

Sources/FoundationEssentials/Subprocess/Subprocess+Configuration.swift

@@ -201,13 +201,13 @@ extension Subprocess {
             return try await withTaskCancellationHandler {
                 return try await withThrowingTaskGroup(of: RunState<R>.self) { group in
                     group.addTask {
-                        let status = monitorProcessTermination(
+                        let status = await monitorProcessTermination(
                             forProcessWithIdentifier: process.processIdentifier)
                         return .monitorChildProcess(status)
                     }
                     group.addTask {
                         do {
-                            let result = try await body(process, .init(fileDescriptor: writeFd))
+                            let result = try await body(process, .init(input: executionInput))
                             try self.cleanup(
                                 process: process,
                                 childSide: false,
@@ -276,7 +276,7 @@ extension Subprocess {
             return try await withTaskCancellationHandler {
                 return try await withThrowingTaskGroup(of: RunState<R>.self) { group in
                     group.addTask {
-                        let status = monitorProcessTermination(
+                        let status = await monitorProcessTermination(
                             forProcessWithIdentifier: process.processIdentifier)
                         return .monitorChildProcess(status)
                     }

Sources/FoundationEssentials/Subprocess/Subprocess+IO.swift

@@ -130,9 +130,9 @@ extension Subprocess {
 
 // MARK: - Execution IO
 extension Subprocess {
-    internal final class ExecutionInput {
-        
-        internal enum Storage {
+    internal final class ExecutionInput: Sendable {
+
+        internal enum Storage: Sendable {
             case noInput(FileDescriptor?)
             case customWrite(FileDescriptor?, FileDescriptor?)
             case fileDescriptor(FileDescriptor?, Bool)
@@ -145,7 +145,7 @@ extension Subprocess {
         }
 
         internal func getReadFileDescriptor() -> FileDescriptor? {
-            return self.storage.withLock { $0
+            return self.storage.withLock {
                 switch $0 {
                 case .noInput(let readFd):
                     return readFd
@@ -196,8 +196,8 @@ extension Subprocess {
                     // The parent fd should have been closed
                     // in the `body` when writer.finish() is called
                     // But in case it isn't call it agian
-                    try readFd?.close()
-                    $0 = .customWrite(nil, writeFd)
+                    try writeFd?.close()
+                    $0 = .customWrite(readFd, nil)
                 }
             }
         }

Sources/FoundationEssentials/Subprocess/Subprocess.swift

@@ -63,57 +63,38 @@ public struct Subprocess: Sendable {
 
 // MARK: - StandardInputWriter
 extension Subprocess {
-    internal actor StandardInputWriterActor {
-        private let fileDescriptor: FileDescriptor
+    @_nonSendable
+    public struct StandardInputWriter {
 
-        internal init(fileDescriptor: FileDescriptor) {
-            self.fileDescriptor = fileDescriptor
-        }
-
-        @discardableResult
-        public func write<S>(_ sequence: S) async throws -> Int where S : Sequence, S.Element == UInt8 {
-            return try self.fileDescriptor.writeAll(sequence)
-        }
-
-        @discardableResult
-        public func write<S: AsyncSequence>(_ asyncSequence: S) async throws -> Int where S.Element == UInt8 {
-            let sequence = try await Array(asyncSequence)
-            return try self.fileDescriptor.writeAll(sequence)
-        }
-
-        public func finish() async throws {
-            try self.fileDescriptor.close()
-        }
-    }
+        private let input: ExecutionInput
 
-    public struct StandardInputWriter: Sendable {
-
-        private let actor: StandardInputWriterActor
-
-        init(fileDescriptor: FileDescriptor) {
-            self.actor = StandardInputWriterActor(fileDescriptor: fileDescriptor)
+        init(input: ExecutionInput) {
+            self.input = input
         }
 
         public func write<S>(_ sequence: S) async throws where S : Sequence, S.Element == UInt8 {
-            try await self.actor.write(sequence)
+            guard let fd: FileDescriptor = self.input.getWriteFileDescriptor() else {
+                fatalError("Attempting to write to a file descriptor that's already closed")
+            }
+            try await fd.write(sequence)
         }
 
         public func write<S>(_ sequence: S) async throws where S : Sequence, S.Element == CChar {
-            try await self.actor.write(sequence.map { UInt8($0) })
+            try await self.write(sequence.map { UInt8($0) })
         }
 
         public func write<S: AsyncSequence>(_ asyncSequence: S) async throws where S.Element == CChar {
             let sequence = try await Array(asyncSequence).map { UInt8($0) }
-            try await self.actor.write(sequence)
+            try await self.write(sequence)
         }
 
         public func write<S: AsyncSequence>(_ asyncSequence: S) async throws where S.Element == UInt8 {
             let sequence = try await Array(asyncSequence)
-            try await self.actor.write(sequence)
+            try await self.write(sequence)
         }
 
         public func finish() async throws {
-            try await self.actor.finish()
+            try self.input.closeParentSide()
         }
     }
 }
@@ -153,23 +134,33 @@ extension Subprocess.Result: Equatable where T : Equatable {}
 
 extension Subprocess.Result: Hashable where T : Hashable {}
 
-extension POSIXError : Swift.Error {}
-
 // MARK: Internal
 extension Subprocess {
     internal enum OutputCapturingState {
         case standardOutputCaptured(Data?)
         case standardErrorCaptured(Data?)
     }
-    
+
+    private func capture(fileDescriptor: FileDescriptor, maxLength: Int) async throws -> Data{
+        let chunkSize: Int = min(Subprocess.readBufferSize, maxLength)
+        var buffer: [UInt8] = []
+        while buffer.count < maxLength {
+            let captured = try await fileDescriptor.read(upToLength: chunkSize)
+            buffer += captured
+            if captured.count < chunkSize {
+                break
+            }
+        }
+        return Data(buffer)
+    }
+
     internal func captureStandardOutput() async throws -> Data? {
         guard let (limit, readFd) = self.executionOutput
             .consumeCollectedFileDescriptor(),
               let readFd = readFd else {
             return nil
         }
-        let captured = try await readFd.read(upToLength: limit)
-        return Data(captured)
+        return try await self.capture(fileDescriptor: readFd, maxLength: limit)
     }
 
     internal func captureStandardError() async throws -> Data? {
@@ -178,8 +169,7 @@ extension Subprocess {
               let readFd = readFd else {
             return nil
         }
-        let captured = try await readFd.read(upToLength: limit)
-        return Data(captured)
+        return try await self.capture(fileDescriptor: readFd, maxLength: limit)
     }
 
     internal func captureIOs() async throws -> (standardOut: Data?, standardError: Data?) {

Sources/TestSupport/TestSupport.swift

@@ -49,7 +49,6 @@ public typealias FloatingPointFormatStyle = Foundation.FloatingPointFormatStyle
 public typealias NumberFormatStyleConfiguration = Foundation.NumberFormatStyleConfiguration
 @available(macOS 12.0, iOS 15.0, tvOS 15.0, watchOS 8.0, *)
 public typealias CurrencyFormatStyleConfiguration = Foundation.CurrencyFormatStyleConfiguration
-public typealias DiscreteFormatStyle = Foundation.DiscreteFormatStyle
 
 @available(FoundationPreview 0.4, *)
 public typealias DiscreteFormatStyle = Foundation.DiscreteFormatStyle
@@ -148,7 +147,6 @@ public typealias FloatingPointFormatStyle = FoundationInternationalization.Float
 public typealias NumberFormatStyleConfiguration = FoundationInternationalization.NumberFormatStyleConfiguration
 @available(macOS 12.0, iOS 15.0, tvOS 15.0, watchOS 8.0, *)
 public typealias CurrencyFormatStyleConfiguration = FoundationInternationalization.CurrencyFormatStyleConfiguration
-public typealias DiscreteFormatStyle = FoundationEssentials.DiscreteFormatStyle
 
 @available(FoundationPreview 0.4, *)
 public typealias DiscreteFormatStyle = FoundationEssentials.DiscreteFormatStyle

Tests/FoundationEssentialsTests/SubprocessTests.swift

@@ -25,13 +25,15 @@ final class SubprocessTests: XCTestCase {
         let result = String(data: ls.standardOutput!, encoding: .utf8)!
         XCTAssert(ls.terminationStatus.isSuccess)
         XCTAssert(!result.isEmpty)
+        print(result)
     }
 
     func testLongText() async throws {
         let cat = try await Subprocess.run(
             executing: .named("cat"),
             arguments: ["/Users/icharleshu/Downloads/PaP.txt"],
-            output: .collect(limit: 1024 * 1024)
+            output: .collect(limit: 1024 * 1024),
+            error: .discard
         )
         print("after")
         print("Result: \(cat.standardOutput?.count ?? -1)")