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

Most programs default the default buffering policy and since you don't have control of how /sbin/ping handles the output one of the pipes might be blocking the FileHandle.AsyncBytes implementation (not sure why). I got this to work with both pipes at the same time by calling .availableData instead to avoid blocking.

import Foundation

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

let process = Process()

process.executableURL = URL(fileURLWithPath: "/sbin/ping")
process.arguments = ["-c", "10", "diariosur.es"]
process.standardOutput = outputPipe
process.standardError = errorPipe

try? process.run()

func processStdOut() async {
    print("stdout start")
    
    while process.isRunning {
        let data = outputPipe.fileHandleForReading.availableData
        if !data.isEmpty {
            if let line = String(data: data, encoding: .utf8)?.trimmingCharacters(in: .whitespacesAndNewlines) {
                print("stdout data: \(line)")
            }
        }
        
    }
    
    print("stdout finished")
}

func processStdErr() async {
    print("stderr start")
    
    while process.isRunning {
        let data = errorPipe.fileHandleForReading.availableData
        if !data.isEmpty {
            if let line = String(data: data, encoding: .utf8)?.trimmingCharacters(in: .whitespacesAndNewlines) {
                print("stderr data: \(line)")
            }
        }
    }
    
    print("stderr finished")
}
    
    await withTaskGroup(of: Void.self) { group in
        group.addTask {
            await processStdErr()
        }
        
        group.addTask {
            await processStdOut()
        }
    }
    
    process.waitUntilExit()
    

Then I get the following (trimmed) output:

stderr start
stdout start
stdout data: PING diariosur.es (23.213.41.6): 56 data bytes
64 bytes from 23.213.41.6: icmp_seq=0 ttl=57 time=7.060 ms
stdout data: 64 bytes from 23.213.41.6: icmp_seq=1 ttl=57 time=6.562 ms
...
stdout data: 64 bytes from 23.213.41.6: icmp_seq=9 ttl=57 time=7.904 ms
stdout data: --- diariosur.es ping statistics ---
10 packets transmitted, 10 packets received, 0.0% packet loss
round-trip min/avg/max/stddev = 6.562/7.327/9.439/0.783 ms
stdout finished
stderr finished

I tried with curl that defaults to stderr and it is not blocking either:

process.executableURL = URL(fileURLWithPath: "/usr/bin/curl")
process.arguments = ["-N", "--output", "test.zsync", "http://ubuntu.mirror.digitalpacific.com.au/releases/23.04/ubuntu-23.04-desktop-amd64.iso.zsync"]

Edit:

Tested with the following C program:

#include <stdio.h>
#include <unistd.h>

int main() {
    for (int i = 1; i <= 100; ++i) {
        fprintf(stdout, "stdout: %d\n", i); 
        fflush(stdout); 
    
        if (i % 10 == 0) {
                fprintf(stderr, "stderr: %d\n", i); 
                fflush(stderr);
        }   
        usleep(100000);
    }   
    
    return 0;
}

It returns the following output:

stdout start
stderr start
stdout data: stdout: 1
stdout data: stdout: 2
stdout data: stdout: 3
stdout data: stdout: 4
stdout data: stdout: 5
stdout data: stdout: 6
stdout data: stdout: 7
stdout data: stdout: 8
stdout data: stdout: 9
stderr data: stderr: 10
stdout data: stdout: 10
stdout data: stdout: 11
stdout data: stdout: 12
stdout data: stdout: 13
stdout data: stdout: 14
stdout data: stdout: 15
stdout data: stdout: 16
stdout data: stdout: 17
stdout data: stdout: 18
stdout data: stdout: 19
stderr data: stderr: 20