File Coverage

blib/lib/Test/ParallelSubtest.pm

Criterion	Covered	Total	%
statement	159	178	89.3
branch	37	60	61.6
condition	2	6	33.3
subroutine	27	28	96.4
pod	4	4	100.0
total	229	276	82.9

package Test::ParallelSubtest;
use strict;
use warnings;

our $VERSION = '0.05';
$VERSION = eval $VERSION;

use Test::Builder::Module;
our @ISA    = qw(Test::Builder::Module);
our @EXPORT = qw(bg_subtest_wait bg_subtest max_parallel);

our @_kids;
our $_i_am_a_child = 0;

our $MaxParallel = 4;

use Carp;
use Sub::Prepend ();
use TAP::Parser;
use Test::Builder;
use Test::ParallelSubtest::Capture;

Test::Builder->new->can('subtest') or croak
                      "Need a version of Test::Builder with subtest support";

END { bg_subtest_wait() if @_kids }

Sub::Prepend::prepend(
    'Test::Builder::done_testing' => \&bg_subtest_wait
);
Sub::Prepend::prepend(
    'Test::Builder::subtest' => sub {
        # Wait for kids just before entering a subtest.
        bg_subtest_wait();

        # Wait for kids just before leaving a subtest.
        my $subtests = $_[2];
        $_[2] = sub {
            $subtests->();
            bg_subtest_wait();
        };
    }
);

sub import_extra {
    my ($class, $list) = @_;

    my @other;
    while (@$list >= 2) {
        my $item = shift @$list;
        if ($item eq 'max_parallel') {
            max_parallel(shift @$list);
        }
        else {
            push @other, $item;
        }
    }

    @$list = @other;

    return;
}

sub max_parallel (;$) {
    my $new_value = shift;

    my $old_value = $MaxParallel;
    if (defined $new_value) {
        $new_value =~ /^[0-9]+\z/ or croak "non-numeric max_parallel value";
        $MaxParallel = $new_value;
    }

    return $old_value;
}

sub bg_subtest ($&) {
    my ($name, $subtests) = @_;

    $_i_am_a_child and croak "bg_subtest() called from a child process";

    my $tb = Test::Builder->new;

    if ($MaxParallel < 1) {
        # We've been told not to fork.
        return $tb->subtest($name, $subtests);
    }

    while (@_kids >= $MaxParallel) {
        _wait_for_next_kid();
    }

    my ($read_pipe, $write_pipe, $pid) = _pipe_and_fork();

    if (!defined $pid) {
        # Can't fork, fall back to a subtest() call.
        return $tb->subtest($name, $subtests);
    }

    my $out_fh  = $tb->output;
    my $fail_fh = $tb->failure_output;
    my $todo_fh = $tb->todo_output;

    if ($pid) {
        # parent
        close $write_pipe;
        my @caller = caller();
        push @_kids, {
            Pid      => $pid,
            Ppid     => $$,
            Pipe     => $read_pipe,
            Out      => [$out_fh, $fail_fh, $todo_fh],
            TestName => $name,
            Caller   => "$caller[1] line $caller[2]",
        };
        return;
    }
    else {
        # child
        close $read_pipe;
        local $Test::Builder::Level = $Test::Builder::Level + 1;
        _child($write_pipe, $name, $subtests);
    }
}

sub _child {
    my ($write_pipe, $name, $subtests) = @_;

    $_i_am_a_child = 1;

    # Capture the outputs of the subtest as strings.
    my $outer_out  = '';
    my $outer_fail = '';
    my $outer_todo = '';
    my $tb = Test::Builder->new;
    $tb->output(\$outer_out);
    $tb->failure_output(\$outer_fail);
    $tb->todo_output(\$outer_todo);

    # Capture the outputs of tests within the subtest for replay
    # in the parent.
    my $capture = Test::ParallelSubtest::Capture->new;

    $tb->subtest($name, sub {
        $capture->install($tb);
        $subtests->();
    });

    # Pass all the captured output to the parent process.
    _len_prefixed_writes($write_pipe,
        \$outer_out, \$outer_fail, \$outer_todo, $capture->as_string_ref
    );

    close $write_pipe;

    no warnings 'redefine';
    *Test::Builder::DESTROY = sub {}; # For T::B fork in subtest bug
    exit(0);
}

sub bg_subtest_wait {
    return if $_i_am_a_child;

    _wait_for_next_kid() while @_kids;
}

sub _pipe_and_fork {
    my ($read_pipe, $write_pipe, $pid);

    while (1) {
        my $pipe_ok = pipe $read_pipe, $write_pipe;
        if ($pipe_ok) {
            $pid = fork;
            if (defined $pid) {
                return ($read_pipe, $write_pipe, $pid);
            }
        }

        if (@_kids) {
            # The pipe or fork failure could be due to a resource limit,
            # reap a kid and try again.
            _wait_for_next_kid();
        }
        else {
            # No kids and pipe+fork won't work, give up.
            return;
        }
    }
}

sub _wait_for_next_kid {
    my $kid = shift @_kids;

    return unless $kid and $kid->{Ppid} == $$;

    my $tb = Test::Builder->new;
    my ($out_dest, $fail_dest, $todo_dest) = @{ $kid->{Out} };

    my ($outer_out, $outer_fail, $outer_todo, $inner_capture)
                                      = _len_prefixed_reads($kid->{Pipe}, 4);

    waitpid $kid->{Pid}, 0; # Don't let zombies build up.

    if (!defined $outer_out) {
        my $name = "failed child process for '$kid->{TestName}'";
        _run_test_from_kid_in_parent(0, $name, undef, undef, <
ERROR: bg_subtest "$kid->{TestName}" ($kid->{Caller}) aborted:
       Lost contact with the child process.
END
        return;
    }

    my ($ok, $name, $todo, $skip) = _parse_outer_output_line($$outer_out);

    if (!defined $ok) {
        $name = "parse child output for '$kid->{TestName}'";
        _run_test_from_kid_in_parent(0, $name, undef, undef, <
ERROR: bg_subtest "$kid->{TestName}" ($kid->{Caller}) aborted:
       Parsing failure in Test::ParallelSubtest - cannot parse:
       [$$outer_out]
END
        return;
    }

    if (defined $todo and $Test::Builder::VERSION >= 0.95_02) {
        # Recent Test::Builder redirects the fail output to the todo
        # output for a todo subtest.
        $fail_dest = $todo_dest;
    }

    my $cap = Test::ParallelSubtest::Capture->new($inner_capture);
    if ( ! $cap->replay_writes($out_dest, $fail_dest, $todo_dest) ) {
        $name = "garbled child output for '$name'";
        _run_test_from_kid_in_parent(0, $name, undef, undef, <
ERROR: bg_subtest "$name" ($kid->{Caller}) aborted:
       Garbled captured output from the child process
END
        return;
    }

    _run_test_from_kid_in_parent($ok, $name, $todo, $skip);

    print $fail_dest $$outer_fail;
    print $todo_dest $$outer_todo;
}

sub _run_test_from_kid_in_parent {
    my ($ok, $name, $todo, $skip, $internal_failure) = @_;

    no warnings 'redefine';
    local *Test::Builder::todo    = sub { $todo };
    local *Test::Builder::in_todo = sub { defined $todo };

    my $tb = Test::Builder->new;

    if ($internal_failure) {
        $tb->ok($ok, $name);
        $tb->diag($internal_failure);
    }
    elsif (defined $skip) {
        $tb->skip($skip);
    }
    else {
        _ok_without_diag_output($tb, $ok, $name);
    }
}

sub _ok_without_diag_output {
    my ($tb, $pass, $name) = @_;

    my $discard = '';
    my $save_fail = $tb->failure_output;
    my $save_todo = $tb->todo_output;
    $tb->failure_output(\$discard);
    $tb->todo_output(\$discard);

    $tb->ok($pass, $name);

    $tb->todo_output($save_todo);
    $tb->failure_output($save_fail);
}

sub _parse_outer_output_line {
    my $output = shift;

    while ($output =~ s/^(\s*#\s*.*\n)//) {
        _print_to(Test::Builder->new->output, $1);
    }
    $output =~ s/^\s*//; # it may have been generated within a subtest

    my $parser = TAP::Parser->new( { tap => $output } ) or return;
    my $result = $parser->next or return;
    $result->is_test or return;

    my $ok = $result->is_actual_ok;
    my $name = $result->description;
    $name =~ s/- //;
    $name =~ s/\\#/#/g;
    my $todo = $result->has_todo ? $result->explanation : undef;
    my $skip = $result->has_skip ? $result->explanation : undef;

    return ($ok, $name, $todo, $skip);
}

sub _len_prefixed_writes {
    my ($fh, @data) = @_;

    print $fh map { pack('N', length $$_) . $$_ } @data;
}

sub _len_prefixed_reads {
    my ($fh, $count) = @_;

    my @results;
    while ($count--) {
        my $lenbuf = '';
        read $fh, $lenbuf, 4 or return;
        length($lenbuf) == 4 or return;
        my $wantlen = unpack 'N', $lenbuf;

        my $buf = '';
        if ($wantlen) {
            read $fh, $buf, $wantlen or return;
            length($buf) == $wantlen or return;
        }
        push @results, \$buf;
    }

    return @results;
}

sub _print_to {
    my ($dest, $msg) = @_;

    if (ref $dest =~ /^SCALAR/) {
        $$dest .= $msg;
    } else {
        print {$dest} $msg;
    }
}

1;

__END__