File Coverage

blib/lib/Parallel/Queue.pm

Criterion	Covered	Total	%
statement	48	48	100.0
branch	19	24	79.1
condition	10	11	90.9
subroutine	9	9	100.0
pod	0	2	0.0
total	86	94	91.4

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

package Parallel::Queue v4.0.0;
use v5.24;
use mro qw( c3 );

use mro::EVERY;

use Carp            qw( croak                       );
use Symbol          qw( qualify_to_ref              );

use Scalar::Util
qw
(
    blessed
    reftype
    looks_like_number
);

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

our @CARP_NOT   = ( __PACKAGE__, qw( mro mro::EVERY ) );

# config state

my %defaultz =
(
    fork    => ! $^P
  , qw
    (
        export  runqueue
        finish  0
        debug   0
        verbose 0
    )
);

my %argz   = %defaultz;

########################################################################
# utility subs
########################################################################

my $format
= sub
{
    state $dumper
    = do
    {
        require Data::Dumper;
        Data::Dumper->can( 'Dumper' )
    };

    local $Data::Dumper::Terse      = 1;
    local $Data::Dumper::Indent     = 1;
    local $Data::Dumper::Sortkeys   = 1;

    local $Data::Dumper::Purity     = 0;
    local $Data::Dumper::Deepcopy   = 0;
    local $Data::Dumper::Quotekeys  = 0;

    my $head    = shift;

    say join "\n" =>
    map
    {
        ref $_
        ? $_->$dumper
        : $_
    }
    ( "($$) $head" => @_ );

    return;
};

my $stub    = sub{};
my $debug   = $stub;

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

my $next_job
= sub
{
    state $redo     = __SUB__;
    state $object   = '';

    my $job     = '';

    if( $object )
    {
        # objects may have their own internal stack.
        # if there aren't any arguments then pass
        # nothing rather than undef from shift.

        if( my $job = $object->next_job )
        {
            return $job;
        }
        else
        {
            $object  = '';
        }
    }

    @_  or return;

    # silently ignore empty slots.

    my $next = shift
    or goto *$redo;

    my $class   = blessed $next;

    if( $class && $class->can( 'next_job' ) )
    {
        $debug->( "New iterator: '$class'" );

        $object = $next;
        goto &$redo
    }
    elsif( 'CODE' eq reftype $next )
    {
        $next
    }
    else
    {
        my $nastygram
        = $format->( 'Bothched queue: un-blessed, non-coderef', $_[0] );

        $argz{ finish }
        or croak $nastygram;

        say STDERR $nastygram;
        goto &$redo
    }
};

my $run_nofork
= sub
{
    # discard the count, iterate the queue without forking.
    shift;

    $debug->( 'Non-forking queue' );

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

        my $exit
        = eval
        {
            $sub->()
            or next
        };

        say STDERR "\nNon-zero exit: $exit, $@\n";

        if( $argz{ finish } )
        {
            say 'Non-zero exit: Continuing queue.';
        }
        else
        {
            say 'Non-zero exit: Aborting queue.';
            last;
        }
    }

    return
};

my $fork_job
= sub
{
    # don't check @_: next_job may have an object
    # returning next jobs w/ an empty queue.

    my $job = &$next_job
    or return;

    if( ( my $pid = fork ) > 0 )
    {
        $debug->( "fork: $pid" );
        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.

        $debug->( "\tExecuting: '$job'" );

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

        # either way, this process needs to exit.

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

        die "Phorkafobia: $!";
    }
};

my $fork_queue
= sub
{
    # count was validated in runqueue.

    my $count   = shift;

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

    $debug->( "Forking initial $count jobs." );

    &$fork_job for 1 .. $count;

    $debug->( "Processing remainder of queue." );

    my $reap_only   = '';

    while( ( my $pid = wait ) > 0 )
    {
        $debug->( "Complete: $pid ($?)" );

        if( $? )
        {
            # 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.

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

            my $result
            = ( $reap_only = ! $argz{ finish } )
            ? "Non-zero exit: Reaping only to complete queue."
            : "Non-zero exit: Continuing queue."
            ;

            say STDERR "\n$failure\n$result\n";
        }

        $reap_only
        or &$fork_job
    }

    return
};

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

sub runqueue
{
    my $count   = $_[0];

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

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

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

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

    @_
}

sub configure
{
    # discard the current patckage
    shift if $_[0] eq __PACKAGE__;

    %argz
    = map
    {
        my ( $arg, $val ) = split /=/, $_, 2;

        $val    //= 1;

        $val    = !$val
        if $arg =~ s{^ no}{}x;

        ( $arg => $val )
    }
    @_;

    @argz{ qw( fork verbose ) } = ( '', 1 )
    if delete $argz{ debug };

    for( $argz{ export } )
    {
        # numeric for true gets default name.
        # name for true gets whatever's there.

        $_  or next;

        # default does the right thing.

        looks_like_number $_
        and delete $argz{ export };

        m{ \W }x
        and croak "Botched export: '$_' contains non-word chars.";
    }

    while( my($k,$v) = each %defaultz )
    {
        $argz{ $k } //= $v;
    }

    $debug
    = $argz{ verbose }
    ? $format
    : $stub
    ;

    $debug->( 'Configuration:', \%argz );

    if( $argz{ fork } && $^P )
    {
        say STDERR
        'Debugger forking. Check TERM=xterm or $DB::debug_TTY.';
    }

    return
}

sub import
{
    &configure;

    if( my $export = $argz{ export } )
    {
        my $caller  = caller;
        my $ref     = qualify_to_ref $export => $caller;

        undef &{ *$ref };

        $debug->( "Installing $export -> $caller" );

        *$ref       = \&runqueue
    }
}

# keep require happy

1

__END__