File Coverage

blib/lib/Parallel/Supervisor.pm

Criterion	Covered	Total	%
statement	100	139	71.9
branch	21	48	43.7
condition	1	3	33.3
subroutine	20	24	83.3
pod	14	18	77.7
total	156	232	67.2

package Parallel::Supervisor;
# A module for managing children / forked processes
# used in a supervisor / worker pattern, eg Parallel::ForkManager

use strict;
use warnings;
use Carp qw(cluck carp);
use Symbol qw(geniosym);
use IO::Pipe;
use IO::Handle;

our $VERSION = "0.03";

sub new # create a new collection of processes
{
    my $class = shift;
    my $self = {
    # data
    'STRUCTS' => {}, # children ready to run (idx name)
    'PROCESSES' => {}, # children running (idx pid)
    'FINISHED' => {}, # children finished running (idx name)
    'NAMES' => {} # index on running child names (idx pid)
    };

    bless($self, $class);
    return $self;
}

#### Getters for dereferencing Object's data structures
sub structs # returns hash of all the jobs prepared but not attached
{ my $self = shift; return $self->{STRUCTS} }

sub processes # returns hash of all the jobs attached
{ my $self = shift; return $self->{PROCESSES} }

sub names # returns a hash of all pid->STRUCTS of attached processes (%NAMES)
{ my $self = shift; return $self->{NAMES} }

sub finished # returns a hash of all pid->STRUCTS of completed processes (%FINISHED)
{ my $self = shift; return $self->{FINISHED} }

sub prepare # (name, cmd_to_eval) create a child struct and add it to %structs
{
    my ($self, $name, @cmd_args) = @_;
    my $cmd = "@cmd_args";

    return undef unless defined $name;
    return undef if $self->is_ready($name);

    #### Setup IPC
    my ($child_writer, $parent_reader) = (geniosym, geniosym);
    eval {
        pipe( $parent_reader, $child_writer );
    } or cluck "Failed to create pipe for reading child output.";

    $parent_reader->autoflush(1);
    $child_writer->autoflush(1);

    $self->{STRUCTS}->{$name} = { id => $name,
                         cmd => "$cmd",
                         child_writer => $child_writer,
                         parent_reader => $parent_reader
     };

     return 1;
}

sub is_ready # check whether name is in %structs;
{
    my ($self,$name) =  @_;

    return undef unless (keys(%{$self->{STRUCTS}} ) );
    for ( keys %{$self->{STRUCTS}} ) {
        return 1 if $_ eq $name;
    }
    return undef ;
}

sub is_attached # check for a running process with this name
{
    my ($self,$name) =  @_;

    return undef unless ( $self->{NAMES} );
    return grep {$_ eq $name} (values( %{ $self->{NAMES} } ) );
}

# move from %structs to %processes with given pid, register @name
sub attach # ident, pid
{
    my ($self,$name,$pid) =  @_;

    return undef unless $pid =~ /^\d+$/;
    return undef if ( $self->is_attached( $name ) ) ;
    return undef unless ( $self->is_ready( $name ) );

    $self->{PROCESSES}{$pid} = $self->{STRUCTS}{$name};
    delete $self->{STRUCTS}{$name};
    $self->{NAMES}->{$pid} = $name;

    return $?;
}

# move from %processes to %structs (does not check whether process still exists)
sub detach # pid
{
    my ($self, $pid) = @_;

    if ( (! defined $pid) ||  ($pid !~ /^\d+$/) ) {
        carp "Error! Can't detach non-numeric pid $pid\n";
        return undef;
    }

    # TODO : verify actual running process with pid
    my $name = $self->{PROCESSES}{$pid}{id};
    if (! defined $name ) {
        carp "detach: could not find process record for pid $pid\n";
        return undef;
    }
    if ($self->is_ready($name) ) {
        carp "detach: name $name is ready, not running as $pid\n";
        return undef;
    }
    if ($self->{NAMES}) {
        delete $self->{NAMES}->{$pid};
    }

    $self->{FINISHED}{$name} = $self->{PROCESSES}{$pid};
    delete ${$self->{PROCESSES}}{$pid};
    return 1;
}

# delete the struct
sub forget # ident
{
    my ($self, $name) = @_;

    return undef unless defined $name;
    if ($self->is_attached($name) ) {
        carp "Cannot forget $name because it is alive.";
        return undef;
    }
    delete $self->{STRUCTS}->{$name};
    delete $self->{FINISHED}->{$name};
    return 1;
}

sub reset # delete all STRUCTS, PROCESSES, NAMES, FINISHED
{
    my $self = shift;
    $self->{STRUCTS} = {}; # children ready to run (idx name)
    $self->{PROCESSES} = {}; # children running (idx pid)
    $self->{FINISHED} = {}; # children finished running (idx n ame)
    $self->{NAMES} = {}; # index on running child names (idx pid)
}
# NOTE: an all numeric name could be confused with a PID!!!!
#  since both structs and processes are checked for a match
sub get_child # return the hash for child with this name

{
    my ($self, $name) = @_;

    my %ret = ();
    return \%ret unless defined $name;
    if ( $self->is_ready( $name ) ) {
        # print STDERR "get_child: $name is ready\n";
        %ret = $self->{STRUCTS}->{$name};
    }
    if ( $self->is_attached( $name ) ) {
        # print STDERR "get_child: $name is alive\n";
        %ret = $self->{PROCESSES}->{$name};
        $ret{ACTIVE} = 1;
        return \%ret;
    }
    # print STDERR "Could not find the process $name\n";
    return \%ret;
}

# avoid iterating through %processes just to look at each id
sub get_names # return array of alive processes by name
{
    my $self = shift;
    my @ret = ();
    if ($self->{NAMES}) {
        @ret = values(%{$self->{NAMES}} );
    }
    return \@ret;
}

sub get_pids # returns a list with pid of all alive processes
{
    my $self = shift;
    my @ret = keys %{$self->{PROCESSES}};
    return wantarray ? @ret : \@ret; #cheating a bit here
}

# NOTE: will retrun fh refs to the read end of the pipe
#       for all processes which have been attached or detached
#       check keys($self->processes) to see if any children are running
sub get_readers # hash of name=>parent_reader_ IO handles
{
    my $self = shift;
    my %handles = ();
    while ( my ($k,$v) = each(%{$self->{PROCESSES}} ) ) {
        $handles{$v->{id} } = $v->{parent_reader};
    }
    while ( my ($k,$v) = each(%{$self->{FINISHED}} ) ) {
        $handles{$v->{id} } = $v->{parent_reader};
    }
    return \%handles;
}

sub get_all_ready # return list of ready STRUCTS
{
    my $self = shift;
    my %ret = ();
    return \%ret unless ($self->{STRUCTS}) ;
    %ret = %{ $self->{STRUCTS} };
    # print STDERR "get_all_ready found ". keys(%ret) . " elements\n";
    return \%ret;
}

# exercise caution when looping through children:
# this can be used to test if child is acivated, otherwise leads to infinite loop
sub get_next_ready # returns a hash of a ready child process, or an empty hash
{
    my $self = shift;
    my %ret = ();
    return \%ret unless defined $self->{STRUCTS};
    return \%ret unless (keys(%{$self->{STRUCTS}} ) );
    my @ids = keys(%{$self->{STRUCTS}} );
    # ensure sort order enforced so caller can plan ahead
    @ids = sort @ids;

    return \%ret unless defined $ids[0];
    my $idx = $ids[0];
    %ret = %{$self->{STRUCTS}{$idx}} ;
    return \%ret;
}

# methods after =cut are processed by the autosplit program and only compiled on demand


1;

__END__