File Coverage

blib/lib/MultiProcFactory.pm

Criterion	Covered	Total	%
statement	200	221	90.5
branch	50	70	71.4
condition	9	18	50.0
subroutine	34	39	87.1
pod	22	23	95.6
total	315	371	84.9

package MultiProcFactory;
# @(#) $Name:  $ $Id: MultiProcFactory.pm,v 1.6 2004/09/21 23:06:49 aaron Exp $
## Aaron Dancygier

## Base class forking object for distributed processing
## among N children.  Parent aggregates when children are done.

use 5.005;
use strict;
use Carp;
use IO::File;
use IPC::Shareable;

use vars qw($VERSION);
$VERSION = '0.04';

sub catch_int {
  IPC::Shareable->clean_up;
}

sub factory {
  my $baseclass = shift;
  my (%params) = @_;

  my $obj;

  croak("must have work_by parameter defined cannot create class\n")
    unless (defined($params{work_by}));

  my $class = $params{work_by};

  unless ($class =~ /^\w+(?:\:\:\w+)*$/) {
    croak("must supply work_by parameter with class name\n");
  }

  eval "use $class";

  if ($@) {
    croak "Error in factory method\n@_, $@";
  }

  return "$class"->new(%params);
}

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

  my $tmp_log_name = $0;

  $tmp_log_name =~ s/\.(?:[^\.]+)$//;

  $args{log_file} ||= "$tmp_log_name.log";
  $args{log_children} = 0 unless(defined($args{log_children}));
  $args{log_parent} = 1 unless(defined($args{log_parent}));
  $args{log_child_append} ||= 0;
  $args{log_parent_append} ||= 0;

  ## set parent signal handlers
  $SIG{QUIT} = $SIG{ABRT} = $SIG{TERM} = $SIG{INT} = \&catch_int;

  ## set up default signal handlers
  _set_parent_signals(\%args);

  ## set up default shared data structures
  unless ($args{IPC_OFF}) {
    my $scalar_handle = tie my $shm_scalar, 'IPC::Shareable', undef, {destroy => 1};
    my $hash_handle = tie my %shm_hash, 'IPC::Shareable', undef, {destroy => 1};

    $args{share_scalar}{handle} = $scalar_handle;
    $args{share_scalar}{var} =  \$shm_scalar;
    $args{share_hash}{handle} = $hash_handle;
    $args{share_hash}{var} = \%shm_hash;
  }

  croak("required parameter do_child code reference is missing or not a code reference\n")
    unless (ref($args{do_child}) eq 'CODE');

  croak("required parameter do_parent_final code reference is missing or not a code reference\n")
    unless (ref($args{do_parent_final}) eq 'CODE');

  if (ref($args{do_parent_init})) {
    unless (ref($args{do_parent_init}) eq 'CODE') {
      croak("optional parameter do_parent_init must be a code reference\n")
    }
  } else {
    $args{do_parent_init} = sub { return 1; };
  }

  $args{max_proc} ||= 20;

  my $self = bless \%args, $class;

  $self->_set_do_parent_init();
  $self->_set_do_parent_final();
  $self->_set_do_child();
  $self->init();

  return $self;
}

sub run {
  my $self = shift;

  my $pfh;

  $self->{log_file} = $self->set_parent_logname();

  my $mode = ($self->{log_parent_append})
    ? '>>'
    : '>'
  ;

  if ($self->{log_parent}) {
    $self->{logp} = IO::File->new("$mode$self->{log_file}") ||
      croak("unable to open parent filehandle $!\n");
    $pfh = $self->{logp};
    $pfh->autoflush(1);
  }

  $self->do_parent_init();

  if ($self->{log_parent}) {
    $self->log_parent(localtime() . " [$$] parent begin\n");
  }

  my ($pid, $prockey);

  my $max_proc = $self->{max_proc};

  my $child_count = 0;

  my @partition_keys = $self->get_prockeys();

  my $proc_count = 0;

  START: while(@partition_keys) {
    while ($child_count < $max_proc and @partition_keys) {
      my $key = shift @partition_keys;
      $pid = fork();
      $child_count ++;
      $proc_count ++;

      if ($pid) {
        ## parent go get another one
        next;
      } elsif(defined($pid)) {
        local $^W = undef;
	$SIG{QUIT} = $SIG{ABRT} = $SIG{TERM} = $SIG{INT} = undef;
        $self->_set_child_signals();
        $prockey = $key;
        last;
      }
    }

    if ($pid) {
      my $kid;

      do {
        $kid = waitpid(-1, 0);

        if ($self->{log_parent}) {
	  $self->log_parent(localtime() . " [$$] child_count: $child_count, parent repeaping: $kid\n");
        }

        $child_count --;

        next START;
      } until ($kid == -1);

       next;
    }

    if ($self->{log_children}) {
      my $tmp_log_name = $self->{log_file};
      $tmp_log_name =~ s/\.log$//;
      $self->{current_child_logname} = "$tmp_log_name\_$proc_count\.log";
      $self->{current_child_logname} = $self->set_child_logname();

      my $mode = ($self->{log_child_append})
        ? '>>'
	: '>'
      ;

      $self->{logc} = IO::File->new("$mode$self->{current_child_logname}") ||
        croak("unable to open child filehandle n child [$$]$!\n");
      $self->{logc}->autoflush(1);
    }

    $self->{prockey} = $prockey;
    $self->do_child_init();
    $self->work();

    if ($self->{log_children}) {
      $self->{logc}->close();
    }

    exit(0);
  }

  my $kid;

  do {
    $kid = waitpid(-1, 0);

    if ($self->{log_parent}) {
      $self->log_parent(localtime() . " [$$] child_count: $child_count, parent repeaping: $kid\n");
    }

    $child_count --;

  } until ($kid == -1);

  $self->do_parent_final();

  if ($self->{log_parent}) {
    $self->log_parent(localtime() . " [$$] parent done\n");
    $pfh->close();
  }

  return 1;
}

sub init {
  my $self = shift;

  croak("init() must be implemented in $self->{'work_by'}\n");
}

sub do_child_init {
  my $self = shift;

  croak("do_child_init() must be implemented in $self->{'work_by'}\n");
}

sub work {
  my $self = shift;

  croak("work() must be implemented in $self->{'work_by'}\n");
}

sub log_parent {
  my ($self, $text) = @_;

  if ($self->{log_parent}) {
    my $pfh = $self->{logp};

    print $pfh $text;
  }
}

sub log_child {
  my ($self, $text) = @_;

  if ($self->{log_children}) {
    my $cfh = $self->{logc};

    print $cfh $text;
  }
}

sub get_prockey {
  my $self = shift;

  return $self->{prockey};
}

sub get_prockeys {
  my $self = shift;

  return (keys %{$self->{partition_hash}});
}

sub scalar_lock {
  my $self = shift;

  (! $self->{IPC_OFF})
    ? return $self->{share_scalar}{handle}->shlock()
    : return undef
  ;
}

sub scalar_unlock {
  my $self = shift;

  (! $self->{IPC_OFF})
    ? return $self->{share_scalar}{handle}->shunlock()
    : return undef
  ;
}

sub hash_lock {
  my $self = shift;

  (! $self->{IPC_OFF})
    ? return $self->{share_hash}{handle}->shlock()
    : return undef
  ;
}

sub hash_unlock {
  my $self = shift;

  (! $self->{IPC_OFF})
    ? return $self->{share_hash}{handle}->shunlock()
    : return undef
  ;
}

sub set_hash_element {
  my $self = shift;

  my ($key, $value) = @_;

  if (! $self->{IPC_OFF}) {
    $self->hash_lock();
    $self->{share_hash}{var}{$key} = $value;
    $self->hash_unlock();
  }
}

sub get_hash_element {
  my ($self, $key) = @_;

  (! $self->{IPC_OFF})
    ?  return $self->{share_hash}{var}{$key}
    : return undef
  ;
}

sub set_scalar {
  my $self = shift;

  my $value = shift;

  if (! $self->{IPC_OFF}) {
    $self->scalar_lock();
    ${$self->{share_scalar}{var}} = $value;
    $self->scalar_unlock();
  }
}

sub inc_scalar {
  my $self = shift;

  if (! $self->{IPC_OFF}) {
    $self->scalar_lock();
    ${$self->{share_scalar}{var}} ++;
    $self->scalar_unlock();
  }
}

sub dec_scalar {
  my $self = shift;

  if (! $self->{IPC_OFF}) {
    $self->scalar_lock();
    ${$self->{share_scalar}{var}} --;
    $self->scalar_unlock();
  }
}

sub get_scalar {
  my $self = shift;

  (! $self->{IPC_OFF})
    ? return ${$self->{share_scalar}{var}}
    : return undef
  ;
}

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

  $self->{log_file} .= '.log'
  unless ($self->{log_file} =~ /\.log$/);
  return $self->{log_file};
}

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

  return $self->{current_child_logname};
}

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

  {
    local $^W = undef;
    no strict;
    *{ ref($self) . '::' . 'do_child' } = $self->{do_child};
  };
}

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

  {
    local $^W = undef;
    no strict;
    *{ ref($self) . '::' . 'do_parent_final' } = $self->{do_parent_final};
  };
}

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

  {
    local $^W = undef;
    no strict;
    *{ ref($self) . '::' . 'do_parent_init' } = $self->{do_parent_init};
  };
}

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

  foreach my $signame (keys %{$self->{parent_sig}}) {
    $SIG{$signame} = $self->{parent_sig}{$signame};
  }
}

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

  foreach my $signame (keys %{$self->{child_sig}}) {
    $SIG{$signame} = $self->{child_sig}{$signame};
  }
}

1;

__END__