#!/usr/bin/perl -w
use strict;
use IO::Select;

use constant READ_SIZE => 8192;

my @program = qw(cat /etc/passwd);
my $select = IO::Select->new();
my %collect_line;

for (1..4) {
    open(my $fh, "-|", @program) || die "Could not start @program: $!";
    $select->add($fh);
    $collect_line{$fh} = "";
}

sub line {
    my $line = shift;
    # Do your per line processing here
    print "Process $line";
}

while (%collect_line) {
    for my $fh ($select->can_read) {
        my $rc = sysread($fh, $collect_line{$fh}, 
                         READ_SIZE, length($collect_line{$fh}));
        die "Read error $! from pipe ??" if !defined($rc);
        if ($rc) {
            line($1) while $collect_line{$fh} =~ s/^(.*\n)//;
        } else {
            # EOF
            line($collect_line{$fh}) if $collect_line{$fh} ne "";
            delete $collect_line{$fh};
            $select->remove($fh);
            close($fh);
            die "Unexpected $? returncode from @program\n" if $?;
        }
    }
}