File Coverage

blib/lib/Parallel/Fork/BossWorker.pm

Criterion	Covered	Total	%
statement	116	127	91.3
branch	16	26	61.5
condition	7	14	50.0
subroutine	17	18	94.4
pod	3	7	42.8
total	159	192	82.8

package Parallel::Fork::BossWorker;
#
# $Id: BossWorker.pm 11 2011-07-16 15:49:45Z twilde $
#

use 5.008008;
use strict;
use warnings;
use Carp;
use Data::Dumper qw(Dumper);
use IO::Handle;
use IO::Select;

# Perl module variables
our @ISA = qw();
our $VERSION = '0.05';

sub new {
    my $class = shift;
    my %values = @_;

    my $self = {
        result_handler => $values{result_handler} || undef,  # Method for handling output of the workers
        worker_count   => $values{worker_count}   || 10,     # Number of workers
        global_timeout => $values{global_timeout} || 0,      # Number of seconds before the worker terminates the job, 0 for unlimited
        work_handler   => $values{work_handler},             # Handler which will process the data from the boss
        work_queue     => [],
        msg_delimiter  => $values{msg_delimiter} || "\0\0\0",
        select         => IO::Select->new(),
    };
    $self->{msg_delimiter_length} = length($self->{msg_delimiter});
    bless $self, ref($class) || $class;

    # The work handler is required
    if (not defined $self->{work_handler}) {
        croak("Parameters \`work_handler' is required.");
    }

    return $self;
}

sub add_work(\@) {
    my $self = shift;
    my $work = shift;
    unshift (@{ $self->{work_queue} }, $work);
}

sub process {
    my $self = shift;
    my $handler = shift;

    eval {

        # If a worker dies, there's a problem
        local $SIG{CHLD} = sub {
            my $pid = wait();
            if (defined $self->{workers}->{$pid}) {
                confess("Worker $pid died.");
            }
        };

        # Start the workers
        $self->start();

        # Read from the workers, loop until they all shut down
        while (%{$self->{workers}}) {
            while (my @ready = $self->{select}->can_read()) {
                foreach my $fh (@ready) {
                    my $result = $self->receive($fh);
                    if (!$result) {
                        $self->{select}->remove($fh);
                        print STDERR "$fh got eof\n";
                        next;
                    }

                    # Process the result handler
                    if ($result->{data} && defined $self->{result_handler}) {
                        &{ $self->{result_handler} }( $result->{data} );
                    }

                    # If there's still work to be done, send it to the worker, otherwise shut it down
                    if ($#{ $self->{work_queue} } > -1) {
                        my $worker = $self->{workers}->{$result->{pid}};
                        $self->send(
                            $self->{workers}->{ $result->{pid} }, # Worker's pipe
                            pop(@{ $self->{work_queue} })
                        );
                    } else {
                        $self->{select}->remove($fh);
                        my $fh = $self->{workers}->{ $result->{pid} };
                        delete($self->{workers}->{ $result->{pid} });
                        close($fh);
                    }
                }
            }
        }

        # Wait for our children so the process table won't fill up
        while ((my $pid = wait()) != -1) { }
    };

    if ($@) {
        croak($@);
    }
}

sub start {
    my $self = shift();

    # Create a pipe for the workers to communicate to the boss

    # Create the workers
    foreach (1..$self->{worker_count}) {

        # Open a pipe for the worker
        pipe(my $from_boss, my $to_worker);
        pipe(my $from_worker, my $to_boss);

        # Fork off a worker
        my $pid = fork();

        if ($pid > 0) {

            # Boss
            $self->{workers}->{$pid} = $to_worker;
            $self->{from_worker}->{$pid} = $from_worker;
            $self->{select}->add($from_worker);

            # Close unused pipes
            close($to_boss);
            close($from_boss);

        } elsif ($pid == 0) {

            # Worker

            # Close unused pipes
            close($from_worker);
            close($to_worker);

            # Setup communication pipes
            $self->{to_boss} = $to_boss;
            open(STDIN, '/dev/null');

            # Send the initial request
            $self->send($to_boss, {pid => $$});

            # Start processing
            $self->worker($from_boss);

            # When the worker subroutine completes, exit
            exit;
        } else {
            confess("Failed to fork: $!");
        }
    }
}

sub worker(\*) {
    my $self = shift();
    my $from_boss = shift();

    # Read instructions from the server
    while (my $instructions = $self->receive($from_boss)) {

        # If the handler's children die, that's not our business
        $SIG{CHLD} = 'IGNORE';

        # Execute the handler with the given instructions
        my $result;
        eval {
            # Handle alarms
            local $SIG{ALRM} = sub {
                die "Work handler timed out."
            };

            # Set alarm
            alarm($self->{global_timeout});

            # Execute the handler and get it's result
            if (defined $self->{work_handler}) {
                $result = &{ $self->{work_handler} }($instructions);
            }

            # Disable alarm
            alarm(0);
        };

        # Warn on errors
        if ($@) {
            croak("Worker $$ error: $@");
        }

        # Send the result to the server
        $self->send($self->{to_boss}, {pid => $$, data => $result});
    }
}

sub receive(\*) {
    my $self = shift();

    # Get the file handle
    my $fh = shift();

    # Get a value from the file handle
    my $value;
    my $char;
    while (read($fh, $char, 1)) {
        $value .= $char;
        if (substr($value, -($self->{msg_delimiter_length})) eq $self->{msg_delimiter}) {
            $value = substr($value, 0, -($self->{msg_delimiter_length}));
            last;
        }
    }

    # Deserialize the data
    no strict;
    no warnings;
    my $data = eval($value);

    if ($@) {
        print STDERR "Value: '$value'\n" if $ENV{PFBW_DEBUG};
        confess("Failed to deserialize data: $@");
    }

    return $data;
}

sub send(\*$) {
    my $self = shift();

    # Get the file handle
    my $fh = shift();

    # Get the value which will be sent
    my $value = shift();

    # Print the value to the file handle
    local $Data::Dumper::Deepcopy = 1;
    local $Data::Dumper::Indent = 0;
    local $Data::Dumper::Purity = 1;
    print $fh Dumper($value) . $self->{msg_delimiter};

    # Force the file handle to flush
    $fh->flush();
}

1;
__END__