File Coverage

blib/lib/Parallel/Queue.pm

Criterion	Covered	Total	%
statement	96	110	87.2
branch	60	90	66.6
condition	10	14	71.4
subroutine	11	12	91.6
pod	0	6	0.0
total	177	232	76.2

########################################################################
# housekeeping
########################################################################

package Parallel::Queue;
use v5.12;

use strict;

use Carp            qw( croak                       );
use Scalar::Util    qw( blessed looks_like_number   );
use Symbol          qw( qualify_to_ref              );

########################################################################
# package variables
########################################################################

our $VERSION    = '3.6';
$VERSION = eval $VERSION;

# defaults.

my  $fork       = '';
my  $verbose    = '';
my  $finish     = '';

########################################################################
# execution handlers
########################################################################

sub next_job
{
    state $next = '';
    my $job     = '';

    JOB:
    for(;;)
    {
        if( $next )
        {
            $job    = $next->( $_[0] )
            and last;

            say STDERR "Completed iterator: '$_[0]'"
            if $verbose;

            $next   = '';
            shift;
        }

        @_ or last;

        for my $item ( $_[0] )
        {
            if
            (
                blessed $item
                and
                $next = $item->can( 'next_job' )
            )
            {
                say STDERR "New iterator: '$next' ($_[0])"
                if $verbose;

                next JOB
            }
            elsif( ref $item )
            {
                $job    = shift;
            }
            elsif( $item )
            {
                # these might end up being processing directives
                # later on, for now ignore them.

                say STDERR "Discarding non-queue '$item'";

                shift
            }
            else
            {
                # silently ignore empty filler.

                shift;
            }
        }

        last if $job;
    }

    say STDERR "Next job: '$job'"
    if $verbose;

    $job
    or return
}

sub run_nofork
{
    # discard the count, iterate the queue without forking.

    shift;

    say STDERR "Non-forking queue"
    if $verbose;

    while( my $sub = &next_job )
    {
        # these should all exit zero.

        my $result  = eval { $sub->() };

        say STDERR "Complete: '$result' ($@)"
        if $verbose;

        if( $result )
        {
            say STDERR "Non-zero exit: $result, aborting queue";

            last
        }
        elsif( $@ )
        {
            say STDERR "Error in job: $@";

            last
        }
    }

    return
}

sub fork_job
{
    @_
    or do
    {
        say STDERR "Queue empty."
        if $verbose;

        return
    };

    my $job = &next_job
    or return;

    if( ( my $pid = fork ) > 0 )
    {
        say STDERR "fork: $pid"
        if $verbose;

        # nothing useful to hand back.

        return
    }
    elsif( defined $pid )
    {
        # child passes the exit status of the perl sub call
        # to the caller as our exit status. the O/S will deal
        # with signal values.
        #
        # aside: failing to exit here will cause runaway
        # phorkatosis.

        say STDERR "\tExecuting: '$job'"
        if $verbose;

        my $exitval = eval { $job->() };

        $@
        ? die $@
        : exit $exitval
    }
    else
    {
        # pass back the fork failure for the caller to deal with.

        die "Phorkafobia: $!";
    }
};

sub fork_queue
{
    # count was validated in runqueue.

    my $count = shift;

    # what's left on the stack are the jobs to run.
    # which may be none.
    # if so, we're done.

    say STDERR "Forking initial $count jobs"
    if $verbose;

    &fork_job for 1 .. $count;

    while( (my $pid = wait) > 0 )
    {
        say STDERR "Complete: $pid ($?)"
        if $verbose;

        # this assumes normal *NIX 16-bit exit values,
        # with a status in the high byte and signum
        # in the lower. notice that $status is not
        # masked to 8 bits, however. this allows us to
        # deal with non-zero exits on > 16-bit systems.
        #
        # caller can trap the signals.

        if( $? )
        {
            # bad news, boss...

            my $message
            = do
            {
                if( my $exit = $? >> 8 )
                {
                    "exit( $exit ) by $pid"
                }
                elsif( my $signal = $? & 0xFF )
                {
                    "kill SIG-$signal on $pid"
                }
            };

            $finish
            ? warn $message
            : die  $message
            ;
        }

        # kick off another job if the queue is not empty.

        @_ and &fork_job;
    }

    return
};

# debug or zero count run the jobs without forking,
# simplifies most debugging issues.

sub runqueue
{
    my ( $count ) = @_;

    looks_like_number $count
    or croak "Bogus runqueue: '$count' non-numeric";

    $count < 0
    and croak "Bogus runqueue: negative count ($count)";

    $fork && $count
    ? eval { &fork_queue }
    : eval { &run_nofork }
    ;

    # return the unused portion.
    # this includes any incomplete iterators.

    @_
}

sub import
{
    # discard the current package name and deal
    # with the args. empty arg for 'export'
    # indicates that runqueue needs to be exported.

    my $caller = caller;

    shift;

    @_ or unshift @_,  qw( export );

    my $export  = 1;
    my $subname = 'runqueue';

    $fork       = ! $^P;
    $verbose    = '';
    $finish     = '';

    for my $arg ( @_ )
    {
        my( $name, $value ) = split /=/, $arg;

        $value  //= 1;

        $value = ! $value
        if $name =~ s/^no//;

        if( 'fork' eq $name )
        {
            $fork       = $value;
        }
        elsif( 'verbose' eq $name )
        {
            $verbose    = $value;
        }
        elsif( 'finish' eq $name )
        {
            $finish     = $value;
        }
        elsif( 'debug' eq $name )
        {
            if( $value )
            {
                $fork       = '';
                $verbose    = 1;
            }
        }
        elsif( 'export' eq $name )
        {
            $export = !! $value;

            looks_like_number $value
            or $subname = $value;
        }
        else
        {
            warn "Unknown argument: '$arg' ignored";
        }
    }

    if( $fork && $^P && ! $DB::fork_TTY )
    {
        say STDERR
        'Debugger forking without $DB::fork_TTY; expect problems';
    }

    if( $export )
    {
        my $ref = qualify_to_ref $subname, $caller;

        undef &{ *$ref };

        *$ref   = \&runqueue
    }

    return
}

sub configure
{
    @_ and import @_, qw( noexport );
}

# keep require happy

1

__END__