File Coverage

blib/lib/Parallel/Tiny.pm

Criterion	Covered	Total	%
statement	75	76	98.6
branch	12	18	66.6
condition	7	14	50.0
subroutine	16	17	94.1
pod	2	2	100.0
total	112	127	88.1

package Parallel::Tiny;

use strict;
use warnings;
use 5.12.0;

use POSIX qw(WNOHANG);
use Sys::Prctl qw(prctl);

## prctl() defaults
use constant PR_SET_PDEATHSIG => 1;
use constant SIGHUP           => 1;

## new() defaults
use constant DEFAULT_ERROR_TIMEOUT => 10;
use constant DEFAULT_REAP_TIMEOUT  => .1;
use constant DEFAULT_SUBNAME       => 'run';
use constant DEFAULT_WORKERS       => 1;
use constant DEFAULT_WORKER_TOTAL  => 1;

our $VERSION = 0.20;

=head1 NAME

Parallel::Tiny

=head1 DESCRIPTION

Provides a very simple, no frills fork manager.

=head1 SYNOPSIS

    my $obj = My::Handler->new();

    my $forker = Parallel::Tiny->new(
        handler      => $obj,
        subname      => 'start', # My::Handler must implement start()
        workers      => 4,
        worker_total => 'infinite',
    );

    $forker->run();

=head1 METHODS

=over

=item new()

Returns a new Parallel::Tiny fork manager.

takes arguments as a hash or hashref with the following arguments:

  handler - an object you provide which has a run() method (unless you define "subname")
            (required)

  reap_timeout - the number of seconds to wait between runs of
            waitpid() to reap children
            (default .1)

  subname - the name of the sub you want to invoke on child spawn
            (default 'run')

  workers - the number of simoltaneous forked processes you
            want to allow at one time
            (default 1)

  worker_total - the total number of processes that you want to run
            (default 1)

You can for instance, say that you want to run 100 proccesses,
but only 4 at a time like this:

    my $forker = Parallel::Tiny->new(
        handler      => $obj,
        workers      => 4,
        worker_total => 100,
    );

If you want you can provide 'infinite' for worker_total.
If you do this, you're responsible for stopping the fork manager!

Signals:
---

If the parent is sent SIGTERM, the parent will wait to reap all children.

If the parent is killed before its children finish, children are configured to receive HUP.

=cut

sub new {
    my $class = shift;
    my $args  = ref($_[0]) ? $_[0] : {@_};

    # set some defaults
    $args->{reap_timeout} ||= DEFAULT_REAP_TIMEOUT;
    $args->{subname}      ||= DEFAULT_SUBNAME;
    $args->{workers}      ||= DEFAULT_WORKERS;
    $args->{worker_total} ||= DEFAULT_WORKER_TOTAL;

    # special configuration
    undef $args->{worker_total} if $args->{worker_total} eq 'infinite';

    # check args
    die 'no handler provided' unless $args->{handler};
    die "handler doesn't implement $args->{subname}()" unless $args->{handler}->can($args->{subname});

    return bless({
            _continue     => 1,
            _handler      => $args->{handler},
            _jobs         => {},
            _reap_timeout => $args->{reap_timeout},
            _subname      => $args->{subname},
            _workers      => $args->{workers},
            _worker_total => $args->{worker_total},
    }, $class);
}

=item run()

Start spooling jobs according to the configuration.

=cut

sub run {
    my $self = shift;

    local $SIG{TERM} = sub { $self->{_continue} = 0 };

    # setup the fork manager
    my $handler = $self->{_handler};
    my $subname = $self->{_subname};

    while ($self->_waitqueue()) {
        # parent work
        my $pid = fork();
        if ($pid) {
            $self->{_worker_total}-- if defined $self->{_worker_total} and $self->{_worker_total} > 0;
            $self->{_jobs}{$pid} = 1;
            next;
        }

        # child work
        prctl(PR_SET_PDEATHSIG, SIGHUP);
        $SIG{$_} = 'DEFAULT' for keys(%SIG);
        $0 = $0 . ' - worker';
        $handler->$subname();

        # child cleanup
        exit 0;
    }

    # wait for children
    while ( wait() != -1 ) {}

    return 1;
}

## Private
############

## waits for another job slot to become available
## short circuits if we've received SIGTERM or reached worker total threshold
sub _waitqueue {
    my $self = shift;

    # check for any stopping conditions
    return 0 if defined $self->{_worker_total} and $self->{_worker_total} <= 0;

    # wait to reap at least one child
    while (keys(%{ $self->{_jobs} }) >= $self->{_workers}) {
        return 0 unless $self->{_continue};
        $self->_reapchildren();
        sleep $self->{_reap_timeout};
    }

    return 1;
}

## checks all currently running jobs and reaps any that have finished, opening
## their slot.
sub _reapchildren {
    my $self = shift;
    foreach my $pid (keys(%{ $self->{_jobs} })) {
        my $waitpid = waitpid($pid, WNOHANG);
        delete $self->{_jobs}{$pid} if $waitpid > 0;
    }
}

=back

=cut

1;

# ABSTRACT: Provides a very simple, no frills fork manager.