How do I intercept the unbuffered output of a Proc::Async in Raku?

Question

With a snippet like

# Contents of ./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"            }
}

executed like

./run raku -e 'react whenever Supply.interval: 1 { .say }'

I expected to see something like

PID: 1234
OUT: 0
OUT: 1
OUT: 2
OUT: 3
OUT: 4
Done

but instead I see

PID: 1234
OUT: 0
Done

I understand that this has to do with buffering: if I change that command into something like

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

I get the desired output.

I know that TTY IO::Handle objects are unbuffered, and that in this case the $*OUT of the spawned process is not one. And I've read that IO::Pipe objects are buffered "so that a write without a read doesn't immediately block" (although I cannot say I entirely understand what this means).

But no matter what I've tried, I cannot get the unbuffered output stream of a Proc::Async. How do I do this?

I've tried binding an open IO::Handle using $proc.bind-stdout but I still get the same issue.

Note that doing something like $proc.bind-stdout: $*OUT does work, in the sense that the Proc::Async object no longer buffers, but it's also not a solution to my problem, because I cannot tap into the output before it goes out. It does suggest to me that if I can bind the Proc::Async to an unbuffered handle, it should do the right thing. But I haven't been able to get that to work either.

---

For clarification: as suggested with the Perl example, I know I can fix this by disabling the buffering on the command I'll be passing as input, but I'm looking for a way to do this from the side that creates the Proc::Async object.

Answer 1

You can set the .out-buffer of a handle (such as $*OUT or $*ERR) to 0:

$ ./run raku -e '$*OUT.out-buffer = 0; react whenever Supply.interval: 1 { .say }'

PID: 11340
OUT: 0
OUT: 1
OUT: 2
OUT: 3
OUT: 4
Done

Answer 2

Proc::Async itself isn't performing buffering on the received data. However, spawned processes may do their own depending on what they are outputting to, and that's what is being observed here.

Many programs make decisions about their output buffering (among other things, such as whether to emit color codes) based on whether the output handle is attached to a TTY (a terminal). The assumption is that a TTY means a human is going to be watching the output, and thus latency is preferable to throughput, so buffering is disabled (or restricted to line buffering). If, on the other hand, the output is going to a pipe or a file, then the assumption is that latency is not so important, and buffering is used to achieve a significant throughput win (a lot less system calls to write data).

When we spawn something with Proc::Async, the standard output of the spawned process is bound to a pipe - which is not a TTY. Thus the invoked program may use this to decide to apply output buffering.

If you're willing to have another dependency, then you can invoke the program via. something that fakes up a TTY, such as unbuffer (part of the expect package, it seems). Here's an example of a program that is suffering from buffering:

my $proc = Proc::Async.new: 'raku', '-e',
    'react whenever Supply.interval(1) { .say }';
react whenever $proc.stdout {
    .print
}

We only see a 0 and then have to wait a long time for more output. Running it via unbuffer:

my $proc = Proc::Async.new: 'unbuffer', 'raku', '-e',
    'react whenever Supply.interval(1) { .say }';
react whenever $proc.stdout {
    .print
}

Means that we see a number output every second.

Could Raku provide a built-in solution to this some day? Yes - by doing the "magic" that unbuffer itself does (I presume allocating a pty - kind of a fake TTY). This isn't trivial - although it is being explored by the libuv developers; at least so far as Rakudo on MoarVM goes, the moment there's a libuv release available offering such a feature, we'll work on exposing it.