poll.pl

#!perl

use strict;
use warnings;

use IPC::Open3;

use Socket;
use IO::Handle;
use IO::Poll;

use POSIX 'WNOHANG';

use Benchmark;

my @tests = qw/test01.t test02.t test03.t test04.t/;

our $max_worker = 2;

our %children;
our %output;

$SIG{CHLD} = \&decrement;
$SIG{PIPE} = 'IGNORE';

my $start = Benchmark->new;
my $poll = IO::Poll->new;

while( 1 ) {

    while ( ( keys %children ) >= $max_worker ) {
        select( undef, undef, undef, 5 );
    }

    if ( @tests ) {
        my $test = shift @tests;

        my ( $ch, $pa );
        socketpair( $ch, $pa, AF_UNIX, SOCK_STREAM, PF_UNSPEC )
            or die "socketpair: $!";    

        my $pid = fork;

        die "Can't fork" unless defined $pid;

        if ( $pid ) {
            close $pa;

            $poll->mask( $ch => POLLOUT );
            $children{$pid} = { child => $ch, output => '' };

            print "Forked:$pid\t$test\n";

        } else {
            close $ch;

            $SIG{CHLD} = 'DEFAULT';
            execute_test( $test, $pa );
            close $pa;
            exit;
        }
    }

    if ( $poll->poll( 0 ) >= 1 ) {

        my @ready = $poll->handles;
        for my $fh ( @ready ) {

            my $ch_id;
            for my $child ( keys %children ) {
                if ( $children{$child}->{child} == $fh ) {
                    $ch_id = $child;
                    last;
                }
            }

            my $buf;
            while ( $fh->sysread( $buf, 4096 ) ) {
                $children{$ch_id}->{output} .= $buf;
            }

        }

    }

    if ( ( ! @tests ) and ( ! keys %children ) ) {
        print "exited\n";
        my $end = Benchmark->new;

        print timestr( timediff( $end, $start ) ) . "\n";
        exit;
    }

}


sub execute_test {
    my $test = shift;
    my $pa   = shift;

    my $r;
    my $w;
    my $pid = open3 $w, $r, 0, ( 'perl', $test );
    close $w;

    my $output = '';
    my $buf;

    while ( my $line = <$r> ) {
        $output .= $line;
    }

    close $r;
    waitpid( $pid, 0 );

    $pa->syswrite( $output );
}

sub decrement {
    while( ( my $child = waitpid( -1, WNOHANG ) ) > 0 ) {
        my $buf;
        while ( $children{$child}->{child}->sysread( $buf, 4096 ) ) {
            $children{$child}->{output} .= $buf;
        }

        $poll->remove( $children{$child}->{child} );
        close $children{$child}->{child};

        if ( $children{$child}->{output} ) {
            print $children{$child}->{output};
        }

        delete $children{$child};
        print "close child:" . $child . "\n";
    }

}