File Coverage

blib/lib/Parallel/Subs.pm

Criterion	Covered	Total	%
statement	90	127	70.8
branch	19	44	43.1
condition	5	16	31.2
subroutine	13	15	86.6
pod	7	7	100.0
total	134	209	64.1

package Parallel::Subs;
$Parallel::Subs::VERSION = '0.002';
use strict;
use warnings;

use Parallel::ForkManager;
use Sys::Info;

# ABSTRACT: Simple way to run subs in parallel and process their return value in perl


sub new {
    my ( $class, %opts ) = @_;

    my $self = bless {}, __PACKAGE__;

    $self->_init(%opts);

    return $self;
}

sub _init {
    my ( $self, %opts ) = @_;

    $self->_pfork(%opts);
    $self->{result} = {};
    $self->{pfork}->run_on_finish(
        sub {
            my ( $pid, $exit, $id, $exit_signal, $core_dump, $data ) = @_;
            die "Failed to process on one job, stop here !"
              if $exit || $exit_signal;
            $self->{result}->{$id} = $data->{result};
        }
    );
    $self->{jobs}      = [];
    $self->{callbacks} = [];

    return $self;
}

sub _pfork {
    my ( $self, %opts ) = @_;

    my $cpu;
    if ( defined $opts{max_process} ) {
        $cpu = $opts{max_process};
    }
    else {
        my $factor = $opts{max_process_per_cpu} || 1;
        eval { $cpu = Sys::Info->new()->device('CPU')->count() * $factor; };
    }
    if ( defined $opts{max_memory} ) {
        my $free_mem;
        eval {
            require Sys::Statistics::Linux::MemStats;
            $free_mem = Sys::Statistics::Linux::MemStats->new->get->{realfree};
        };
        my $max_mem = $opts{max_memory} * 1024;  # 1024 **2 = 1 GO => expr in Kb
        my $cpu_for_mem;
        if ($@) {

#warn "Cannot guess amount of available free memory need Sys::Statistics::Linux::MemStats\n";
            $cpu_for_mem = 2;
        }
        else {
            $cpu_for_mem = int( $free_mem / $max_mem );
        }

        # min
        $cpu = ( $cpu_for_mem < $cpu ) ? $cpu_for_mem : $cpu;
    }
    $cpu ||= 1;

    $self->{cpu} = $cpu;

    # we could also set a minimum amount of required memory
    $self->{pfork} = Parallel::ForkManager->new( int($cpu) );
    $self->{pfork}->set_waitpid_blocking_sleep(0)
      unless $opts{waitpid_blocking_sleep};

    return $self;
}


sub add {
    my ( $self, $code, $test ) = @_;

    return unless $code && ref $code eq 'CODE';
    push(
        @{ $self->{jobs} },
        { name => ( scalar( @{ $self->{jobs} } ) + 1 ), code => $code }
    );
    push( @{ $self->{callbacks} }, $test );

    return $self;
}


sub total_jobs {
    my ($self) = @_;

    return scalar @{ $self->{jobs} };
}


sub wait_for_all_optimized {
    my ($self) = @_;

    return $self unless $self->total_jobs;
    my @original_jobs = @{ $self->{jobs} };

    # callback not supported for now
    if ( scalar @{ $self->{callbacks} } ) {
        warn "Callback not supported in this mode for now.\n"
          if grep { defined $_ } @{ $self->{callbacks} };
        $self->{callbacks} = [];
    }

    my $cpu = $self->{cpu} or die;
    my $jobs_per_cpu = int( scalar @original_jobs / $cpu );
    ++$jobs_per_cpu if scalar @original_jobs % $cpu || !$jobs_per_cpu;

    my @new_jobs;

    my $generate_sub = sub {
        my ( $from, $to ) = @_;

        return sub {

            #print "subprocess from $from to $to\n";
            for ( my $i = $from ; $i <= $to ; ++$i ) {

                #print "running job $i\n";
                $original_jobs[$i]->{code}->();
            }
            return;
        };
    };

    my ( $from, $to ) = ( 0, 0 );
    foreach my $id ( 1 .. $cpu ) {
        $to = $from + $jobs_per_cpu - 1;
        $to = scalar(@original_jobs) - 1 if $to >= scalar(@original_jobs);

        #print "FROM $from - TO $to\n";
        my $sub = $generate_sub->( $from, $to );

        push @new_jobs, { name => $id, code => $sub };
        $from = $to + 1;
    }

    $self->{jobs} = \@new_jobs;

    return $self->wait_for_all();
}


sub run {
    my ($self) = @_;

    return unless scalar @{ $self->{jobs} };
    my $pfm = $self->{pfork};
    for my $job ( @{ $self->{jobs} } ) {
        $pfm->start( $job->{name} ) and next;
        my $job_result = $job->{code}();

        # can be used to stop on first error
        my $job_error = 0;
        $pfm->finish( $job_error, { result => $job_result } );
    }

    # wait for all jobs
    $pfm->wait_all_children;

    return $self->{result};
}


sub wait_for_all {
    my ($self) = @_;

    # run callbacks
    die "Cannot run callbacks" unless $self->run();

    return $self unless $self->total_jobs;
    my $c = 0;

    my $results = $self->results();

    foreach my $callback ( @{ $self->{callbacks} } ) {
        next unless $callback;
        die "cannot find result for #${c}" unless exists $results->[$c];
        my $res = $results->[ $c++ ];

        if ( ref $callback eq 'HASH' ) {

            # internal mechanism
            return
              unless defined $callback->{test} && defined $callback->{args};

            my @args = ( $res, @{ $callback->{args} } );
            my $t    = $callback->{test};
            my $str  = join( ', ', map { "\$args[$_]" } ( 0 .. $#args ) );
            eval "$t(" . $str . ")";
        }
        elsif ( ref $callback eq 'CODE' ) {

            # execute user function
            $callback->($res);
        }

    }

    return $self;
}


sub results {
    my ($self) = @_;

    my @sorted =
      map  { $self->{result}{$_} }
      sort { int($a) <=> int($b) } keys %{ $self->{result} };
    return \@sorted;
}

1;

__END__