File Coverage

blib/lib/Metabrik/Worker/Parallel.pm

Criterion	Covered	Total	%
statement	9	49	18.3
branch	0	14	0.0
condition	0	3	0.0
subroutine	3	9	33.3
pod	2	6	33.3
total	14	81	17.2

#
# $Id$
#
# worker::parallel Brik
#
package Metabrik::Worker::Parallel;
use strict;
use warnings;

use base qw(Metabrik);

sub brik_properties {
   return {
      revision => '$Revision$',
      tags => [ qw(unstable) ],
      author => 'GomoR ',
      license => 'http://opensource.org/licenses/BSD-3-Clause',
      attributes => {
         pool_size => [ qw(count) ],
         pid => [ qw(pid) ],  # pid is available within son process only
         manager => [ qw(INTERNAL) ],
      },
      attributes_default => {
         pool_size => 10,
      },
      commands => {
         create_manager => [ qw(pool_size|OPTIONAL) ],
         reset_manager => [ ],
         start => [ qw(sub) ],
         stop => [ ],
      },
      require_modules => {
         'Parallel::ForkManager' => [ ],
      },
   };
}

sub create_manager {
   my $self = shift;
   my ($pool_size) = @_;

   # Do not create another manager if one already exists.
   my $manager = $self->manager;
   if (defined($manager)) {
      return 1;
   }

   $pool_size ||= $self->pool_size;

   $manager = Parallel::ForkManager->new(
      $pool_size,
   );

   $self->manager($manager);

   return 1;
}

sub reset_manager {
   my $self = shift;

   my $manager = $self->manager;
   if (! defined($manager)) {
      return 1;
   }

   $manager->wait_all_children;
   $self->manager(undef);

   return 1;
}

sub start {
   my $self = shift;
   my ($sub) = @_;

   $self->brik_help_run_undef_arg('start', $sub) or return;
   my $ref = $self->brik_help_run_invalid_arg('start', $sub, 'CODE')
      or return;

   $self->create_manager or return;
   my $manager = $self->manager;

   my $pid = $manager->start;
   if ($pid) {  # Success, return to parent
      # This is the child pid, to be used by parent:
      $self->pid($pid);
      return $pid;
   }

   # Continue within son
   $self->pid($$);  # This pid is the child pid, gathered from within child
   &{$sub}();

   $manager->finish;

   return 0;
}

sub stop {
   my $self = shift;

   my $manager = $self->manager;
   if (defined($manager)) {
      $manager->wait_all_children;
   }

   return 1;
}

sub brik_fini {
   my $self = shift;

   $self->reset_manager;

   return $self->SUPER::brik_fini;
}

1;

__END__