read process standardOutput and standardError in parallel in swift without blocking

Question

In swift5 would like to run a Process() read both standardOutput and standardError without blocking, so I can parse them.

This example code once the line with for try await line in errorPipe.fileHandleForReading.bytes.lines is called, the program execution is blocked. The standardOutput reader stops printing

import Foundation

let outputPipe = Pipe()
let errorPipe = Pipe()

let process = Process()
process.executableURL = URL(fileURLWithPath:"/sbin/ping")
process.arguments = ["google.com"]
process.standardOutput = outputPipe
process.standardError = errorPipe

try? process.run()

func processStdOut() async
{
  for i in 0..<5 {
    print("processStdOut X ", i)
    try? await Task.sleep(nanoseconds: 1_000_000_000)
  }

  do {
    for try await line in outputPipe.fileHandleForReading.bytes.lines {
      print("stdout Line: \(line)")
    }
  } catch {
    NSLog("processStdOut Error \(error.localizedDescription)")
  }
  NSLog("processStdOut finished")

}

func processStdErr() async
{
  for i in 0..<5 {
    print("processStdErr X ", i)
    try? await Task.sleep(nanoseconds: 2_000_000_000)
  }
  do {
    for try await line in errorPipe.fileHandleForReading.bytes.lines {
      print("stderr Line: \(line)")
    }
  } catch {
    NSLog("processStdErr Error \(error.localizedDescription)")
  }
  NSLog("processStdErr finished")
}

await withTaskGroup(of: Void.self) { group in
  group.addTask {
    await processStdErr()
  }
  group.addTask {
    await processStdOut()
  }
  group.addTask {
    process.waitUntilExit()
  }
}

Note that if you force data into standardError by disconnecting the wifi or network standardOutput is unblocked again.

Anything else I should try?

Answer 1

Yes, it appears that the standard bytes implementation can block standardOutput when simultaneously using bytes on standardError, too.

Here is a simple bytes implementation that does not block, because it avails itself of readabilityHandler:

extension Pipe {
    struct AsyncBytes: AsyncSequence {
        typealias Element = UInt8

        let pipe: Pipe

        func makeAsyncIterator() -> AsyncStream<Element>.Iterator {
            AsyncStream { continuation in
                pipe.fileHandleForReading.readabilityHandler = { @Sendable handle in
                    let data = handle.availableData

                    guard !data.isEmpty else {
                        continuation.finish()
                        return
                    }

                    for byte in data {
                        continuation.yield(byte)
                    }
                }

                continuation.onTermination = { _ in
                    pipe.fileHandleForReading.readabilityHandler = nil
                }
            }.makeAsyncIterator()
        }
    }

    var bytes: AsyncBytes { AsyncBytes(pipe: self) }
}

Thus, the following does not experience the same problem when simultaneously processing both standardOutput and standardError at the same time:

let outputPipe = Pipe()
let errorPipe = Pipe()

let process = Process()
process.executableURL = URL(fileURLWithPath: …)
process.standardOutput = outputPipe
process.standardError = errorPipe

func processStandardOutput() async throws {
    for try await line in outputPipe.bytes.lines {
        …
    }
}

func processStandardError() async throws {
    for try await line in errorPipe.bytes.lines {
        …
    }
}

// optionally, you might want to return whatever non-zero termination status code the process returned

process.terminationHandler = { process in
    if process.terminationStatus != 0 {
        exit(process.terminationStatus)
    }
}

try process.run()

try? await withThrowingTaskGroup(of: Void.self) { group in
    group.addTask {
        try await processStandardOutput()
    }
    
    group.addTask {
        try await processStandardError()
    }
    
    …
}