如何在Raku中拦截Proc::Async的无缓冲输出？

最近我遇到了一个Proc::Async的输出缓冲问题，折腾了好一阵子，现在把问题和解决方案整理出来，希望能帮到同样踩坑的人。

首先看我的测试脚本（文件./run）：

my $p = Proc::Async.new: @*ARGS;
react {
    whenever Promise.in: 5 { $p.kill }
    whenever $p.stdout { say "OUT: { .chomp }" }
    whenever $p.ready { say "PID: $_" }
    whenever $p.start { say "Done" }
}

当我执行./run raku -e 'react whenever Supply.interval: 1 { .say }'时，预期应该每隔一秒看到一行OUT: n，5秒后结束输出Done，但实际只输出了：

PID: 1234
OUT: 0
Done

这明显是缓冲在搞鬼！我知道像Perl里可以给子进程加$|++禁用缓冲，比如执行这个命令就正常了：

./run perl -E '$|++; while(1) { state $i; say $i++; sleep 1 }'

查了资料才明白：TTY类型的IO::Handle是无缓冲的，但这里子进程的$*OUT不是TTY，而是IO::Pipe——这玩意儿本身是带缓冲的，官方文档提到“这样无读取操作的写入不会立即阻塞”，虽然没完全搞懂这句背后的细节，但缓冲确实是问题根源。

我试过用$proc.bind-stdout绑定自己打开的IO::Handle，但没效果。不过意外发现如果绑定到$*OUT（当前终端的输出句柄），Proc::Async就不缓冲了——但这没法捕获输出内容，只能直接打印，不是我要的方案。这说明如果能绑定一个无缓冲的句柄，应该能解决问题，于是摸索出两个可行的方法：

方法1：自定义无缓冲句柄绑定 #

我们可以手动创建一个无缓冲的IO::Handle，把它绑定到Proc::Async的stdout，然后实时读取这个句柄的内容：

my $unbuffered-handle = IO::Handle.new;
$unbuffered-handle.buffering = False; # 核心：关闭缓冲
$unbuffered-handle.open: :w, :pipe; # 以管道模式打开句柄

my $p = Proc::Async.new: @*ARGS;
$p.bind-stdout: $unbuffered-handle;

react {
    whenever Promise.in: 5 { $p.kill }
    # 从无缓冲句柄读取并处理输出
    whenever Supply.from-handle($unbuffered-handle) -> $chunk {
        for $chunk.lines -> $line {
            say "OUT: { $line.chomp }" if $line.defined;
        }
    }
    whenever $p.ready { say "PID: $_" }
    whenever $p.start {
        $unbuffered-handle.close; # 子进程结束后关闭句柄
        say "Done";
    }
}

方法2：二进制模式读取+手动处理字节流 #

另一种思路是绕过默认的行缓冲，用二进制模式读取stdout，然后自己处理字节流中的换行，确保每一行都能被及时捕获：

my $p = Proc::Async.new: @*ARGS;
my $buffer = Buf.new; # 用来临时存放未处理的字节

react {
    whenever Promise.in: 5 { $p.kill }
    whenever $p.stdout(:bin) -> $buf {
        $buffer ~= $buf;
        # 循环查找换行符，处理每一行
        while $buffer.index("\n".encode) -> $pos {
            my $line = $buffer.substr(0, $pos).decode;
            say "OUT: { $line.chomp }";
            $buffer = $buffer.substr($pos + 1); # 截断已处理的部分
        }
    }
    whenever $p.ready { say "PID: $_" }
    whenever $p.start {
        # 处理子进程结束后剩余的缓冲内容
        if $buffer.bytes {
            say "OUT: { $buffer.decode.chomp }";
        }
        say "Done";
    }
}

原理说明 #

• 方法1通过显式创建无缓冲句柄，让子进程的输出直接写入这个句柄，不会被管道的默认缓冲拦截，所以我们能实时拿到输出。
• 方法2用二进制模式读取，避免了Proc::Async默认的行缓冲机制，手动拆分字节流，确保每一行输出都能被及时处理。

这样修改后，再执行原来的Raku命令，就能得到预期的逐行输出了！

内容的提问来源于stack exchange，提问作者jja