File Coverage

blib/lib/Parallel/ForkManager/Segmented.pm

Criterion	Covered	Total	%
statement	36	36	100.0
branch	9	14	64.2
condition			n/a
subroutine	6	6	100.0
pod	1	1	100.0
total	52	57	91.2

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

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  /};

    my $pm;

    if ($WITH_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} ) )
    {
        if ($WITH_PM)
        {
            my $pid = $pm->start;

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

1;

__END__