File Coverage

blib/lib/Parallel/MPM/Prefork.pm

Criterion	Covered	Total	%
statement	136	251	54.1
branch	37	140	26.4
condition	29	94	30.8
subroutine	23	32	71.8
pod	8	8	100.0
total	233	525	44.3

package Parallel::MPM::Prefork;

use 5.010;
use strict;
use warnings;
use Exporter 'import';
use Fcntl;
use POSIX qw(:signal_h :sys_wait_h sigprocmask);
use Socket;
use Storable qw(nfreeze thaw);

use Data::Dumper;

use constant {
  MAX_SERVERS => 73,
  MAX_SPARE => 10,
  MIN_SPARE => 5,
  START_SERVERS => 5,
};

use constant CLD_DATA_HDR_FMT => 'iSCL'; # PID, exit code, thaw, data length
use constant CLD_DATA_HDR_LEN => length pack CLD_DATA_HDR_FMT, 0;

our $VERSION = '0.14';

our (@EXPORT_OK, @EXPORT_TAGS) = ();
our @EXPORT =
  qw(
      pf_init
      pf_done
      pf_whip_kids
      pf_kid_new
      pf_kid_busy
      pf_kid_yell
      pf_kid_idle
      pf_kid_exit
  );

our $error;

my $pgid;
my $done;
my $debug;
my $am_parent;
my $timeout;

my $max_servers;
my $max_spare_servers;
my $min_spare_servers;
my $start_servers;

my $parent_stat_fh;
my $parent_data_fh;

my $child_stat_fh;
my $child_stat_fd;

my $child_data_fh;
my $child_data_fd;

my $child_fds;

my $child_data_hook;
my $child_sigh;

my $dhook_in_main;
my $dhook_pid;

my $num_busy;
my $num_idle;
my %busy;
my %idle;

my $sigset_bak = POSIX::SigSet->new();
my $sigset_all = POSIX::SigSet->new();
$sigset_all->fillset();

#
# Public interface
#

sub pf_init {
  my %opts = @_;

  eval {
    setpgrp();
    $pgid = getpgrp();

    $timeout = $am_parent = 1;
    $dhook_pid = $done = $num_busy = $num_idle = 0;
    $child_fds = $child_stat_fd = $child_data_fd = $error = '';

    undef %busy;
    undef %idle;

    $debug = $opts{debug};

    # Just like Apache, we allow start_servers to be larger than
    # max_spare_servers to accommodate for high initial load.
    $max_servers = int($opts{max_servers} // MAX_SERVERS);
    $max_spare_servers = int($opts{max_spare_servers} // MAX_SPARE);
    $min_spare_servers = int($opts{min_spare_servers} // MIN_SPARE);
    $start_servers = int($opts{start_servers} // START_SERVERS);

    if ($max_servers <= 0 || $max_spare_servers <= 0 ||
        $min_spare_servers <= 0 || $start_servers <= 0) {
      die "All child server process numbers must be >= 1!";
    }

    if ($max_servers < $min_spare_servers) {
      $max_servers = $min_spare_servers;
      warn "Adjusted max_servers to $max_servers";
    }

    if ($max_spare_servers < $min_spare_servers) {
      $max_spare_servers = $min_spare_servers;
      warn "Adjusted max_spare_servers to $max_spare_servers";
    }

    if ($start_servers > $max_servers) {
      $start_servers = $max_servers;
      warn "Adjusted start_servers to $start_servers";
    }
    elsif ($start_servers < $min_spare_servers) {
      $start_servers = $min_spare_servers;
      warn "Adjusted start_servers to $start_servers";
    }

    if (defined($child_data_hook = $opts{child_data_hook})) {
      ref $child_data_hook eq 'CODE' or
        die "child_data_hook must be a code reference";
    }

    $child_sigh = _make_child_sigh($opts{child_sigh});
    $dhook_in_main = $opts{data_hook_in_main};

    if ($child_data_hook) {
      _socketpair($parent_data_fh, $child_data_fh);
      if ($dhook_in_main) {
        vec($child_data_fd, fileno $child_data_fh, 1) = 1;
        $child_fds |= $child_data_fd;
      }
      else {
        $dhook_pid = _fork_data_hook_helper($parent_data_fh, $child_data_fh);
      }
    }

    _socketpair($parent_stat_fh, $child_stat_fh);
    vec($child_stat_fd, fileno $child_stat_fh, 1) = 1;
    $child_fds |= $child_stat_fd;

    $SIG{CHLD} = \&_sig_chld;
    _wait_for_children()
  };

  if ($@) {
    $error = $@;
    return undef;
  }

  return $parent_stat_fh;
}

sub pf_whip_kids ($;$) {
  my $code = shift;
  my $args = shift;

  return 0 if $done;

  if ($start_servers) {
    do { _spawn($code, $args) } until ! --$start_servers;
  }
  elsif ((my $lack = $min_spare_servers - $num_idle) > 0) {
    _log_child_action('Forking', $lack) if $debug;
    for (1 .. $lack) {
      last if (_spawn($code, $args) // return undef) < 0;
    }
  }
  elsif ((my $plus = $num_idle - $max_spare_servers) > 0) {
    _log_child_action('Killing', $plus) if $debug;
    _kill_idlers($plus);
  }

  _log_child_status() if $debug;
  _read_child_drool();

  return 1;
}

sub pf_kid_new () {
  while (!$done) {
    if ($start_servers) {
      $start_servers--;
      return _spawn();
    }
    elsif ((my $lack = $min_spare_servers - $num_idle) > 0) {
      _log_child_action('Lacking', $lack, "Forking 1 child.\n") if $debug;
      my $pid = _spawn();
      return $pid if !defined $pid || $pid >= 0;
    }
    elsif ((my $plus = $num_idle - $max_spare_servers) > 0) {
      _log_child_action('Killing', $plus) if $debug;
      _kill_idlers($plus);
    }

    _log_child_status() if $debug;
    _read_child_drool();
  }

  return -1;
}

sub pf_kid_busy {
  syswrite $parent_stat_fh, "R$$\n" if ! $am_parent;
}

sub pf_kid_idle {
  syswrite $parent_stat_fh, "S$$\n" if ! $am_parent;
}

sub pf_kid_yell($;$$) {
  my ($data, $thaw, $exitcode) = @_;

  return undef if $am_parent || !($parent_data_fh && ref $data);

  $data = eval { nfreeze($data) } // do {
    warn "ERROR: Could not nfreeze() data from pid $$: ", $@;
    $error = $@;
    return undef;
  };

  syswrite $parent_data_fh,
    pack(CLD_DATA_HDR_FMT, $$, $exitcode // 256, $thaw ? 1 : 0, length $data)
    . $data;
}

sub pf_kid_exit(;$$$) {
  my ($exitcode, $data, $thaw) = @_;

  return if $am_parent;

  ($exitcode //= 0) &= 0xff;

  pf_kid_yell($data, $thaw, $exitcode);

  exit $exitcode;
}

sub pf_done (;$) {
  my $exitcode = shift;

  return if !$am_parent || $done++;

  local $SIG{CHLD} = 'IGNORE';
  local $SIG{TERM} = 'IGNORE';
  kill 'TERM', 0;

  my $pid = 0;
  my $nbytes;

  undef $child_fds if ! $child_stat_fh;

  do {
    $pid = waitpid -$pgid, WNOHANG if $pid >= 0;
    $nbytes = _read_child_data();
    select my $rfds = $child_fds, undef, undef, .1 if !($pid || $nbytes);
  } while $pid >= 0 || $nbytes;

  exit $exitcode if defined $exitcode;

  undef $_ for ($parent_stat_fh, $child_stat_fh,
                $parent_data_fh, $child_data_fh);
}


#
# Private parts
#

sub _make_child_sigh {
  my $child_sigh = shift // return undef;

  ref $child_sigh eq 'HASH' or die "child_sigh must be a hash reference";

  if (%$child_sigh) {
    my %sig2hnd;
    while (my ($sigs, $code) = each %$child_sigh) {
      if (defined $code &&
          ref $code ne 'CODE' && $code !~ /^(?:DEFAULT|IGNORE)$/) {
        die "child_sigh($sigs) must be a code ref, DEFAULT, IGNORE or undef";
      }
      for (split ' ', $sigs) {
        $sig2hnd{$_} =
          exists $SIG{$_} ? $code : die "child_sigh: No such signal: $_";
      }
    }
    return \%sig2hnd;
  }

  return undef;
}

sub _socketpair {
  socketpair $_[0], $_[1], AF_UNIX, SOCK_STREAM, PF_UNSPEC
    or die "ERROR: socketpair(): $!\n";
  fcntl $_[1], F_SETFL, fcntl($_[1], F_GETFL, 0) | O_NONBLOCK;
}

sub _fork_data_hook_helper {
  my ($parent_data_fh, $child_data_fh) = @_;

  sigprocmask(SIG_BLOCK, $sigset_all, $sigset_bak);

  my $cpid = fork() // die "Could not fork: $!";

  if ($cpid) {
    sigprocmask(SIG_SETMASK, $sigset_bak);
    return $cpid;
  }

  undef $parent_data_fh;
  $0 .= ' [data_hook_helper]';

  $child_fds = '';
  vec($child_fds, fileno $child_data_fh, 1) = 1;

  while (my ($sig, $hnd) = each %SIG) {
    undef $SIG{$sig} if defined $hnd && $sig ne 'FPE';
  }

  sigprocmask(SIG_SETMASK, $sigset_bak);

  while (1) {
    select my $rfds = $child_fds, undef, undef, undef;
    _read_child_data();
  }
}

sub _spawn {
  my $code = shift;
  my $args = shift;

  if ($num_idle + $num_busy >= $max_servers) {
    warn "Server seems busy, consider increasing max_servers.\n";
    _log_child_status();
    return -1;
  }

  # Temporarily block signal delivery until child has installed all handlers
  # and knows for sure it's not the parent.
  sigprocmask(SIG_BLOCK, $sigset_all, $sigset_bak);

  my $cpid = fork();

  if ($cpid) {
    # Parent
    $num_idle++;
    $idle{$cpid}++;
  }
  elsif (defined $cpid) {
    # Child
    undef $am_parent;
    undef $child_data_fh;
    undef $child_stat_fh;

    @SIG{keys %$child_sigh} = values %$child_sigh if $child_sigh;

    if ($code) {
      sigprocmask(SIG_SETMASK, $sigset_bak);
      $code->(@{$args // []});
      exit;
    }
  }

  sigprocmask(SIG_SETMASK, $sigset_bak);

  $cpid;
}

sub _sig_chld {
  # force select() to return immediately if child exited shortly before
  $timeout = 0;
}

sub _wait_for_children {
  my $ct;
  while ((my $pid = waitpid -$pgid, WNOHANG) > 0) {
    if ($pid == $dhook_pid) {
      warn "ERROR: data_hook_helper exited, forking new one.\n";
      $dhook_pid = _fork_data_hook_helper($parent_data_fh, $child_data_fh);
    }
    else {
      delete $busy{$pid} and $num_busy--;
      delete $idle{$pid} and $num_idle--;
      warn "PID $pid exited.\n" if $debug;
    }
    $ct++;
  }
  $ct;
}

sub _read_child_drool {
  my $status_changed;
  do {
    if (select my $rfds = $child_fds, undef, undef, $timeout) {
      $status_changed = unpack '%32b*', $rfds & $child_stat_fd;
      if ($dhook_in_main && unpack '%32b*', $rfds & $child_data_fd) {
        _read_child_data();
        $status_changed ||= select $rfds = $child_stat_fd, undef, undef, 0;
      }
      _read_child_status() if $status_changed;
    }
    $timeout = 1;
  } until _wait_for_children() || $status_changed;
}

# An in-memory scoreboard would surely be nicer ...
sub _read_child_status {
  sigprocmask(SIG_BLOCK, $sigset_all, $sigset_bak);
  while (<$child_stat_fh>) {
    my ($status, $pid) = unpack 'aA*';
    # Ignore delayed status messages from no longer existing children
    next unless $busy{$pid} or $idle{$pid};
    if ($status eq 'R') {
      delete $idle{$pid} and $num_idle--;
      $busy{$pid}++ or $num_busy++;
    }
    elsif ($status eq 'S') {
      delete $busy{$pid} and $num_busy--;
      $idle{$pid}++ or $num_idle++;
    }
    elsif ($status ne '0') { # 0 = Jeffries tube. cg use only!
      warn "ERROR: Dubious status: $_";
    }
  }
  sigprocmask(SIG_SETMASK, $sigset_bak);
}

sub _read_child_data {
  return undef unless $child_data_fh && fileno $child_data_fh;

  my $nbytes = 0;
  my $chunks = $dhook_in_main ? 3 : ~0;  # read at most that many chunks per
                                         # call

  sigprocmask(SIG_BLOCK, $sigset_all, $sigset_bak);

 HDR:
  while ($chunks-- && sysread $child_data_fh, my $header, CLD_DATA_HDR_LEN) {
    my ($pid, $exitcode, $thaw, $data_len) = unpack CLD_DATA_HDR_FMT, $header;

    # Exit code 256 means "undef", minimum nfreeze() data length is 3.
    if ($pid <= 1 || $exitcode > 256 || $thaw > 1 || $data_len < 3) {
      warn(
        'ERROR: read corrupted child data: ',
        "pid:$pid exitcode:$exitcode thaw:$thaw data_len:$data_len",
        ', skipping all pending data'
      );
      $nbytes += sysread($child_data_fh, $header, 16384) || last HDR while 1;
    }

    my $cbytes = sysread($child_data_fh, (my $data), $data_len) // do {
      warn "ERROR: sysread(): $!";
      next HDR;
    };

    $nbytes += $cbytes;

    if ($cbytes != $data_len) {
      warn "ERROR: sysread(): read $cbytes bytes but expected $data_len";
      next HDR;
    }

    # Don't block signals in the data hook.
    sigprocmask(SIG_SETMASK, $sigset_bak);
    $child_data_hook->(
      $pid,
      ($thaw ? eval { thaw $data } : $data) //
        do {
          warn "ERROR: Could not thaw() data from pid $pid: ", $@;
          $error = $data;
          undef;
        },
      $exitcode <= 255 ? $exitcode : undef,
    );
    sigprocmask(SIG_BLOCK, $sigset_all, $sigset_bak) if $chunks;
  }

  sigprocmask(SIG_SETMASK, $sigset_bak);

  $nbytes;
}

sub _kill_idlers {
  my $plus = shift;

  $num_idle -= $plus;
  kill 'TERM', my @idlers = (keys %idle)[0 .. --$plus];
  delete @idle{@idlers};
}

sub _log_child_action {
  my ($what, $count, @more) = @_;
  warn "$what $count child", $count == 1 ? ".\n" : "ren.\n", @more;
}

sub _log_child_status {
  warn "busy:$num_busy idle:$num_idle\n";
}

1;

__END__