File Coverage

blib/lib/Parallel/ForkManager/Segmented.pm

Criterion	Covered	Total	%
statement	36	36	100.0
branch	6	8	75.0
condition			n/a
subroutine	7	7	100.0
pod	1	1	100.0
total	50	52	96.1

package Parallel::ForkManager::Segmented;
$Parallel::ForkManager::Segmented::VERSION = '0.10.1';
use strict;
use warnings;
use 5.014;

use Parallel::ForkManager::Segmented::Base v0.4.0 ();
use parent 'Parallel::ForkManager::Segmented::Base';

use Parallel::ForkManager ();

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

    my $processed = $self->process_args($args);
    return if not $processed;
    my ( $WITH_PM, $batch_cb, $batch_size, $nproc, $stream_cb, ) =
        @{$processed}{qw/ WITH_PM batch_cb batch_size nproc stream_cb  /};

    return $self->serial_run($processed) if not $WITH_PM;

    my $pm;

    $pm = Parallel::ForkManager->new($nproc);
    my $batch = $stream_cb->( { size => 1 } )->{items};
    return if not defined $batch;
    $batch_cb->($batch);
ITEMS:
    while (
        defined( $batch = $stream_cb->( { size => $batch_size } )->{items} ) )
    {
        my $pid = $pm->start;

        if ($pid)
        {
            next ITEMS;
        }
        $batch_cb->($batch);
        $pm->finish;    # Terminates the child process
    }
    $pm->wait_all_children;
    return;
}

1;

__END__