File Coverage

blib/lib/Parallel/Fork/BossWorkerAsync.pm

Criterion	Covered	Total	%
statement	299	332	90.0
branch	95	148	64.1
condition	34	50	68.0
subroutine	27	27	100.0
pod	5	18	27.7
total	460	575	80.0

package Parallel::Fork::BossWorkerAsync;
use strict;
use warnings;
use Carp;
use Data::Dumper qw( Dumper );
use Socket       qw( AF_UNIX SOCK_STREAM PF_UNSPEC );
use Fcntl        qw( F_GETFL F_SETFL O_NONBLOCK );
use POSIX        qw( EINTR EWOULDBLOCK );
use IO::Select ();

our @ISA = qw();
our $VERSION = '0.09';

# TO DO (wish list):
# Restart crashed child workers.

# -----------------------------------------------------------------
sub new {
  my ($class, %attrs)=@_;
  my $self = {
    work_handler    => $attrs{work_handler},                # required
    init_handler    => $attrs{init_handler}   || undef,     # optional
    exit_handler    => $attrs{exit_handler}   || undef,     # optional
    result_handler  => $attrs{result_handler} || undef,     # optional
    worker_count    => $attrs{worker_count}   || 3,         # optional, how many child workers
    global_timeout  => $attrs{global_timeout} || 0,         # optional, in seconds, 0 is unlimited
    msg_delimiter   => $attrs{msg_delimiter}  || "\0\0\0",  # optional, may not appear in data
    read_size       => $attrs{read_size}      || 1024*1024, # optional, defaults to 1 MB
    verbose         => $attrs{verbose}        || 0,         # optional, *undocumented*, 0=silence, 1=debug
    shutting_down   => 0,
    force_down      => 0,
    pending         => 0,
    result_stream   => '',
    result_queue    => [],
    job_queue       => [],
  };
  bless($self, ref($class) || $class);

  croak("Parameter 'work_handler' is required") if ! defined($self->{work_handler});

  # Start the "boss" process, which will start the workers
  $self->start_boss();

  return $self;
}

# -----------------------------------------------------------------
sub serialize {
  my ($self, $ref)=@_;
  local $Data::Dumper::Deepcopy = 1;
  local $Data::Dumper::Indent = 0;
  local $Data::Dumper::Purity = 1;
  return Dumper($ref) . $self->{msg_delimiter};
}

# -----------------------------------------------------------------
sub deserialize {
  my ($self, $data)=@_;
  $data = substr($data, 0, - length($self->{msg_delimiter}));
  my $VAR1;
  my $ref = eval($data);
  if ($@) {
    confess("failed to deserialize: $@");
  }
  return $ref;
}

# -----------------------------------------------------------------
# Pass one or more hashrefs for the jobs.
# Main app sends jobs to Boss.
sub add_work {
  my ($self, @jobs)=@_;
  $self->blocking($self->{boss_socket}, 1);
  while (@jobs) {
    $self->log("add_work: adding job to queue\n");
    my $job = shift(@jobs);
    my $n = syswrite( $self->{boss_socket}, $self->serialize($job) );
    croak("add_work: app write to boss: syswrite: $!") if ! defined($n);
    $self->{pending} ++;
    $self->log("add_work: job added to queue, $self->{pending} pending\n");
  }
}

# -----------------------------------------------------------------
# Syntactic nicety
sub get_result_nb {
  my ($self)=@_;
  return $self->get_result(blocking => 0);
}

# -----------------------------------------------------------------
# Main app gets a complete, single result from Boss.
# If defined, result_handler fires here.
# Return is result of work_handler, or result_handler (if defined),
# or {} (empty hash ref).
# Undef is returned if socket marked nonblocking and read would have
# blocked.
sub get_result {
  my ($self, %args)=@_;
  $args{blocking} = 1 if ! defined($args{blocking});
  carp("get_result() when no results pending") if ! $self->pending();

  my $rq_count = scalar(@{ $self->{result_queue} });
  $self->log("get_result: $self->{pending} jobs in process, $rq_count results ready\n");

  if ( ! @{ $self->{result_queue} }) {
    $self->blocking($self->{boss_socket}, $args{blocking});
    $self->read($self->{boss_socket}, $self->{result_queue}, \$self->{result_stream}, 'app');

    # Handle nonblocking case
    if ( ! $args{blocking}  &&  ! @{ $self->{result_queue} }) {
      return undef;
    }
  }

  $self->log("get_result: got result\n");

  $self->{pending} --;
  if ($self->{pending} == 0  &&  $self->{shutting_down}) {
    $self->log("get_result: no jobs pending; closing boss\n");
    close($self->{boss_socket});
  }
  my $ref = $self->deserialize( shift( @{ $self->{result_queue} } ) );
  my $retval = $self->{result_handler} ? $self->{result_handler}->($ref) : $ref;
  $retval = {} if ! defined($retval);
  return $retval;
}

# -----------------------------------------------------------------
# Main app calls to see if there are submitted jobs for which no
# response has been collected.  It doesn't mean the responses are
# ready yet.
sub pending {
  my ($self)=@_;
  return $self->{pending};
}

# -----------------------------------------------------------------
# App tells boss to shut down by half-close.
# Boss then finishes work in progress, and eventually tells
# workers to exit.
# Boss sends all results back to app before exiting itself.
# Note: Boss won't be able to close cleanly if app ignores
# final reads...
# args: force => 0,1  defaults to 0
sub shut_down {
  my ($self, %args)=@_;
  $args{force} ||= 0;
  $self->{shutting_down} = 1;

  $self->log("shut_down: MARK\n");

  if ($args{force}) {
    # kill boss pid
    kill(9, $self->{boss_pid});
  } elsif ($self->pending()) {
    shutdown($self->{boss_socket}, 1);
  } else {
    close($self->{boss_socket});
  }

  while (wait() != -1) {};		# waits/reaps Boss process
}

# -----------------------------------------------------------------
# Make socket blocking/nonblocking
sub blocking {
  my ($self, $socket, $makeblocking)=@_;

  ### --- W I N D O W S --- ###
  if ($^O eq 'MSWin32') {
    # ioctl() requires a pointer to a long, containing the nonblocking value.

    # The long var
    my $nonblocking = pack('L', $makeblocking ? 0 : 1);

    # The pointer to it
    my $plong = unpack('I', pack('P', $nonblocking));

    # The nonblocking request. FIONBIO is 0x8004667e
    ioctl($socket, 0x8004667e, $plong)
      or croak("ioctl failed: $!");


  ### --- LINUX, BSD, etc --- ###
  } else {
    my $flags = fcntl($socket, F_GETFL, 0)
      or croak("fcntl failed: $!");
    my $blocking = ($flags & O_NONBLOCK) == 0;
    if ($blocking  && ! $makeblocking) {
      $flags |= O_NONBLOCK;
    } elsif (! $blocking && $makeblocking) {
      $flags &= ~O_NONBLOCK;
    } else {
      # do nothing
      return;
    }

    fcntl($socket, F_SETFL, $flags)
      or croak("fcntl failed: $!");
  }

  return;
}

# -----------------------------------------------------------------
sub start_boss {
  my ($self)=@_;
  $self->log("start_boss: start\n");
  eval {
    my ($b1, $b2);
    socketpair($b1, $b2, AF_UNIX, SOCK_STREAM, PF_UNSPEC)
      or die("socketpair: $!");

    my $pid = fork();
    defined $pid || confess("fork failed: $!");

    if ($pid) {
      # Application (parent)
      $self->{boss_pid} = $pid;

      # App won't write to, or read from itself.
      close($b2);
      $self->{boss_socket} = $b1;

      $self->log("start_boss: Application: Boss started\n");

    } else {
      # Manager aka Boss (child)
      # Boss won't write to, or read from itself.
      close($b1);

      $self->{app_socket} = $b2;

      # Make nonblocking
      $self->blocking( $self->{app_socket}, 0 );
      open(STDIN, '/dev/null');

      $self->start_workers();
      $self->boss_loop();
      while (wait() != -1) {};			# waits/reaps workers only

      $self->log("start_boss: Boss: exiting\n");
      exit;
    }
  };
  if ($@) {
    croak($@);
  }
}

# -----------------------------------------------------------------
sub start_workers {
  my ($self)=@_;
  $self->log("start_workers: starting $self->{worker_count} workers\n");
  eval {
    for (1 .. $self->{worker_count}) {
      my ($w1, $w2);
      socketpair($w1, $w2, AF_UNIX, SOCK_STREAM, PF_UNSPEC)
        or die("socketpair: $!");

      my $pid = fork();
      defined $pid || confess("fork failed: $!");

      if ($pid) {
        # Boss (parent)
        close($w2);
        $self->{workers}->{ $w1 } = { pid => $pid, socket => $w1 };

        # Make nonblocking
        $self->blocking( $w1, 0 );

      } else {
        # Worker (child)
        close($self->{app_socket});
        delete($self->{workers});
        close($w1);
        $self->{socket} = $w2;
        open(STDIN, '/dev/null');

        $self->worker_loop();
        exit;
      }
    }

    $self->log("start_workers: start workers complete\n");
  };
  if ($@) {
    croak($@);
  }
}

# -----------------------------------------------------------------
# Boss process; have an open socket to the app, and one to each worker.
# Loop select(), checking for read and write on app socket, and read
# on working children, and write on idle children.
# Keep track of idle vs. working children.
# When receive a shutdown order from the app, keep looping until the
# job queue is empty, and all results have been retrieved (all
# children will now be idle.)  Then close the worker sockets.
# They'll be reading, and will notice this and exit.
# Don't deserialize any data.  Just look for the delimiters to know
# we're processing whole records.
#

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

  $self->log("boss_loop: start\n");
  eval {
    # handy
    my $workers = $self->{workers};

    # All workers start out idle
    for my $s (keys(%$workers)) {
      $workers->{ $s }->{idle} = 1;
    }

    while ( 1 ) {
      # When to exit loop?
      #   shutting_down = 1
      #   job_queue empty
      #   all workers idle, and no partial jobs
      #   result_queue empty
      if ($self->{shutting_down}  &&
          ! @{ $self->{job_queue} }  &&
          ! @{ $self->{result_queue} } ) {
        my $busy=0;
        my $partials = 0;
        for my $s (keys(%$workers)) {
          if ( ! $workers->{ $s }->{idle}) {
            $busy ++;
            last;
          } elsif (exists($workers->{ $s }->{partial_job})) {
            $partials ++;
            last;
          }
        }
        if ( ! $busy  &&  ! $partials) {
          # Close all workers
          for my $s (keys(%$workers)) {
            close($workers->{ $s }->{socket});
          }
          close($self->{app_socket});
          last;
        }
      }

      # Set up selectors:
      # Always check app for read, unless shutting down.  App write only if
      # there's something in @result_queue.
      my (@rpids, @wpids);
      my $rs = IO::Select->new();
      if ( ! $self->{shutting_down}) {
        $rs->add($self->{app_socket});
        push(@rpids, "app");
      }
      my $ws = IO::Select->new();
      if ( @{ $self->{result_queue} } ) {
        $ws->add($self->{app_socket});
        push(@wpids, "app");
      }

      # Check workers for read only if not idle
      # Otherwise, IF job_queue isn't empty,
      # check nonidle workers for write.
      for my $s (keys(%$workers)) {
        if ( $workers->{ $s }->{idle}) {
          if ( @{ $self->{job_queue} }  ||  exists($workers->{ $s }->{partial_job})) {
            $ws->add($workers->{ $s }->{socket});
            push(@wpids, $workers->{ $s }->{pid});
          }
        } else {
          $rs->add($workers->{ $s }->{socket});
          push(@rpids, $workers->{ $s }->{pid});
        }
      }

      # Blocking
      my @rdy = IO::Select->select($rs, $ws, undef);
      if ( ! @rdy) {
        if ($! == EINTR) {
          # signal interrupt, continue waiting
          next;
        }
        croak("select failed: $!");
      }
      my ($r, $w) = @rdy[0,1];

      # Now we have zero or more reabable sockets, and
      # zero or more writable sockets, but there's at
      # least one socket among the two groups.
      # Read first, as things read can be further handled
      # by writables immediately afterwards.

      for my $rh (@$r) {
        my ($source, $queue, $rstream);
        if ($rh != $self->{app_socket}) {
          $source = $workers->{$rh}->{pid};
          $queue = $self->{result_queue};
          $rstream = \$workers->{$rh}->{result_stream};
        } else {
          $source = 'app';
          $queue = $self->{job_queue};
          $rstream = \$self->{job_stream};
        }

        $self->log("boss_loop: reading socket\n");
        $self->read($rh, $queue, $rstream, 'boss');
        $self->log("boss_loop: read socket complete\n");
      }

      for my $wh (@$w) {
        my $source = exists($workers->{ $wh }) ? $workers->{ $wh }->{pid} : "app";
        $self->log("boss_loop: writing socket\n");
        $self->write($wh);
        $self->log("boss_loop: write socket complete\n");
      }
    }
  };
  if ($@) {
    croak($@);
  }
}

# -----------------------------------------------------------------
sub write {
  my ($self, $socket)=@_;
  if ($socket == $self->{app_socket}) {
    $self->write_app($socket);
  } else {
    $self->write_worker($socket);
  }
}

# -----------------------------------------------------------------
sub write_app {
  my ($self, $socket)=@_;

  # App socket: write all bytes until would block, or complete.
  # This means process result_queue in order, doing as many elems
  # as possible.  Don't remove from the queue until complete.  In
  # other words, the first item on the queue may be a partial from
  # the previous write attempt.
  my $queue = $self->{result_queue};
  while (@$queue) {
    $self->log("write_app: processing queue entry\n");
    while ( $queue->[0] ) {
      my $n = syswrite($socket, $queue->[0]);
      if ( ! defined($n)) {
        # Block or real socket error
        if ($! == EWOULDBLOCK) {
          # That's it for this socket, try another, or select again.
          return;
        } else {
          croak("boss write to app: syswrite: $!");
        }
      }

      elsif ($n == 0) {
        # Application error: socket has been closed prematurely by other party.
        # Boss is supposed to close app socket before app.  App tells Boss to
        # stop, but it only happens after all existing work is completed, and
        # data is sent back to app.
        croak("boss write to app: peer closed prematurely");

      } else {
        # wrote some bytes, remove them from the queue elem
        substr($queue->[0], 0, $n) = '';
      }
    }
    # queue elem is empty, remove it, go try next one
    $self->log("write_app: process queue entry complete\n");
    shift(@$queue);
  }
  $self->log("write_app: all queue entries have been written\n");
  # queue is empty, all written!
}

# -----------------------------------------------------------------
sub write_worker {
  my ($self, $socket)=@_;

  # A worker: check to see if we have a remaining partial
  # job we already started to send.  If so, continue with this.
  # Otherwise, take a *single* job off the job_queue, and send that.
  # When we've gotten either complete, or would block, write remaining
  # portion to per-worker job-in-progress, or make it '' if complete.
  # With worker, we only send ONE job, never more.
  # Once job send is complete, mark worker not-idle.

  if ( ! exists($self->{workers}->{ $socket }->{partial_job})) {
    $self->log("write_worker: processing new job\n");
    if (@{ $self->{job_queue} }) {
      $self->{workers}->{ $socket }->{partial_job} = shift(@{ $self->{job_queue} });
    } else {
      # Nothing left on queue.  Remember, we select on *all* idle workers,
      # even if there's only one job on the queue.
      return;
    }
  } else {
    $self->log("write_worker: processing job remnant\n");
  }
  my $rjob = \$self->{workers}->{ $socket }->{partial_job};

  while ( length($$rjob) ) {
    $self->log("write_worker: writing...\n");
    my $n = syswrite($socket, $$rjob);
    if ( ! defined($n)) {
      # Block or real socket error
      if ($! == EWOULDBLOCK) {
        # That's it for this socket, try another, or select again.
        return;
      } else {
        croak("boss write to worker: syswrite: $!");
      }
    }

    elsif ($n == 0) {
      # Application error: socket has been closed prematurely by other party.
      # Boss is supposed to close worker socket before worker - that's how
      # worker knows to exit.
      croak("boss write to worker: peer closed prematurely (pid " . $self->{workers}->{ $socket }->{pid} . ")");

    } else {
      # wrote some bytes, remove them from the job
      substr($$rjob, 0, $n) = '';
      $self->log("write_worker: wrote $n bytes\n");
    }
  }
  # job all written!
  $self->log("write_worker: job complete\n");
  delete($self->{workers}->{ $socket }->{partial_job});
  $self->{workers}->{ $socket }->{idle} = 0;
}

# -----------------------------------------------------------------
# Boss exits loop on error, wouldblock, or shutdown msg (socket close).
# Worker exits loop on error, recd full record, or boss socket close.
# App exits loop on error, recd full record, wouldblock (nb only), early boss close (error).
# Stream (as external ref) isn't needed for worker, as it's blocking, and only reads a single
# record, no more.
# So $rstream can be undef, and if so, we init locally.
sub read {
  my ($self, $socket, $queue, $rstream, $iam)=@_;
  my $stream;
  $rstream = \$stream if ! defined($rstream);
  $$rstream = '' if ! defined($$rstream);

  # croak messaging details...
  my $source;
  if ($iam eq 'boss') {
    if ($socket == $self->{app_socket}) {
      $source = 'app';
    } else {
      $source = "worker [$self->{workers}->{$socket}->{pid}]";
    }
  } else {    # app or worker, same source
    $source = "boss";
  }

  while ( 1 ) {
    $self->log("read: $iam is reading...\n");

    my $n = sysread($socket, $$rstream, $self->{read_size}, length($$rstream));
    if ( ! defined($n)) {
      if ($! == EINTR) {
        # signal interrupt, continue reading
        next;
      } elsif ($! == EWOULDBLOCK) {
        last;    # No bytes recd, no need to chunk.
      } else {
        croak("$iam read from $source: sysread: $!");
      }

    } elsif ($n == 0) {
      # Application error: socket has been closed prematurely by other party.
      # Boss is supposed to close worker socket before worker - that's how
      # worker knows to exit.
      # Boss is supposed to close app socket before app.  App tells Boss to
      # stop, but it only happens after all existing work is completed, and
      # data is sent back to app.
      if ($iam eq 'boss') {
        if ($socket == $self->{app_socket}) {
          $self->{shutting_down} = 1;
        } elsif (exists($self->{workers}->{$socket})) {
          croak("$iam read from $source: peer closed prematurely (pid " . $self->{workers}->{ $socket }->{pid} . ")");
        }
      } elsif ($iam eq 'worker') {
        close($socket);
      } else {    # i am app
        croak("$iam read from $source: peer closed prematurely (pid " . $self->{boss_pid} . ")");
      }

      # if we didn't croak...
      last;

    } else {
      # We actually read some bytes.  See if we can chunk
      # out any record(s).
      $self->log("read: $iam read $n bytes\n");

      # Split on delimiter
      my @records = split(/(?<=$self->{msg_delimiter})/, $$rstream);

      # All but last elem are full records
      my $rcount=$#records;
      push(@$queue, @records[0..$#records-1]);

      # Deal with last elem, which may or may not be full record
      if ($records[ $#records ] =~ /$self->{msg_delimiter}$/) {
        # We have a full record
        $rcount++;
        $self->log("read: $iam pushing full record onto queue\n");
        push(@$queue, $records[ $#records ]);
        $$rstream = '';
        if (exists($self->{workers}->{ $socket })) {
          $self->{workers}->{ $socket }->{idle} = 1;
        }
      } else {
        $$rstream = $records[$#records];
      }

      # Boss grabs all it can get, only exiting loop on wouldblock.
      # App (even nb method), and workers all exit when one full
      # record is received.
      last if $rcount  &&  $iam ne 'boss';
    }
  }
}

# -----------------------------------------------------------------
# Worker process; single blocking socket open to boss.
# Blocking select loop:
# Only do read OR write, not both.  We never want more than a single
# job at a time.  So, if no job currently, read, waiting for one.
# Get a job, perform it, and try to write results.
# Send delimiter, which tells boss it has all the results, and we're ready
# for another job.
#
sub worker_loop {
  my ($self)=@_;
  eval {
    if ($self->{init_handler}) {
      $self->log("worker_loop: calling init_handler()\n");
      $self->{init_handler}->();
    }

    # String buffers to store serialized data: in and out.
    my $result_stream;
    while ( 1 ) {
      if (defined($result_stream)) {
        # We have a result: write it to boss
        $self->log("worker_loop: writing result...\n");

        my $n = syswrite( $self->{socket}, $result_stream);
        croak("worker [$$] write to boss: syswrite: $!") if ! defined($n);
        $self->log("worker_loop: wrote $n bytes\n")       if defined($n);
        $result_stream = undef;
        # will return to top of loop

      } else {
        # Get job from boss

        my @queue;
        $self->log("worker_loop: reading job from queue...\n");
        $self->read($self->{socket}, \@queue, undef, 'worker');
        return if ! @queue;
        $self->log("worker_loop: read job complete, we have a job\n");

        my $job = $self->deserialize($queue[0]);
        my $result;
        eval {
          local $SIG{ALRM} = sub {
            die("BossWorkerAsync: timed out");
          };

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

          # Invoke handler and get result
          $self->log("worker_loop: calling work_handler for this job\n");
          $result = $self->{work_handler}->($job);

          # Disable alarm
          alarm(0);
        };

        if ($@) {
          $result = {ERROR => $@};
          $self->log("worker_loop: ERROR: $@\n");
        }

        $result_stream = $self->serialize($result);
      }
    }
  };
  my $errm = $@ || '';
  eval {
  if ($self->{exit_handler}) {
    $self->log("worker_loop: calling exit_handler()\n");
    $self->{exit_handler}->();
  }
  };
  $errm .= "\n$@" if $@;
  if ($errm) {
    croak($errm);
  }
}


# -----------------------------------------------------------------
# IN: log message
# If verbose is enabled, print the message.
sub log {
  my ($self, $msg) = @_;
  print STDERR $msg   if $self->{verbose};
}


1;
__END__