File Coverage

blib/lib/Parallel/ForkManager/Scaled.pm

Criterion	Covered	Total	%
statement	7	9	77.7
branch			n/a
condition			n/a
subroutine	3	3	100.0
pod			n/a
total	10	12	83.3

package Parallel::ForkManager::Scaled;
use Moo;
use namespace::clean;
use Unix::Statgrab;
use List::Util qw( min max );
use Storable qw( freeze thaw );

use 5.010;

our $VERSION = '0.17';

extends 'Parallel::ForkManager';

has hard_min_procs   => ( is => 'rw', lazy => 1, builder => 1 );
has hard_max_procs   => ( is => 'rw', lazy => 1, builder => 1 );
has soft_min_procs   => ( is => 'rw', lazy => 1, builder => 1, trigger => 1 );
has soft_max_procs   => ( is => 'rw', lazy => 1, builder => 1, trigger => 1 );
has initial_procs    => ( is => 'lazy' );
has update_frequency => ( is => 'rw', default => 1 );
has idle_target      => ( is => 'rw', default => 0 );
has idle_threshold   => ( is => 'rw', default => 1 );
has run_on_update    => ( is => 'rw', clearer => 1, predicate => 1 );

has last_update  => ( is => 'rwp', default => sub{ time } );
has _stats_pct   => ( is => 'rw',  clearer => 1, predicate => 1, handles => [ qw( idle ) ] );
has _host_info   => ( is => 'rw',  clearer => 1, predicate => 1, lazy => 1, builder => 1, handles => [ qw( ncpus ) ] );
has _last_stats  => ( is => 'rw',  clearer => 1, predicate => 1, default => sub{ get_cpu_stats } );

has __unstorable => ( is => 'ro', init_arg => undef, default => sub{[qw( _stats_pct _host_info _last_stats )]} );

#
# Once Parallel::ForkManager has converted to Moo (in development)
# this will no longer be necessary. Probably. :)
#
sub FOREIGNBUILDARGS {
    my ($class, @args) = @_;
    my @ret;

    my $args = @args > 1 ? {@args} : $args[0];

    push @ret, 1; # will get changed later in BUILD()
    push @ret, $args->{tempdir} if defined $args->{tempdir};

    @ret;
}

sub BUILD {
    my $self = shift;
    $self->set_max_procs(min($self->soft_max_procs, max($self->soft_min_procs, $self->initial_procs)));
    $self->update_stats_pct;
};

sub _build_hard_min_procs { 1 }
sub _build_hard_max_procs { (shift->ncpus // 1) * 2 }
sub _build_soft_min_procs { shift->hard_min_procs };
sub _build_soft_max_procs { shift->hard_max_procs };
sub _build__host_info     { get_host_info }

# pick a value half way between our soft min and max
sub _build_initial_procs {
    my $self = shift;
    $self->hard_min_procs+int(($self->soft_max_procs-$self->soft_min_procs)/2);
}

# soft min cannot be less than hard min
sub _trigger_soft_min_procs {
    my ($self, $newval) = @_;

    $self->soft_min_procs($self->hard_min_procs)
        if $newval < $self->hard_min_procs;
}

# soft max cannot exceed hard_max
sub _trigger_soft_max_procs {
    my ($self, $newval) = @_;

    $self->soft_max_procs($self->hard_max_procs)
        if $newval > $self->hard_max_procs;
}

sub update_stats_pct {
    my $self = shift;

    my $stats = get_cpu_stats;
    my $pcts  = $stats->get_cpu_stats_diff($self->_last_stats)->get_cpu_percents;

    # Not enough time has elapsed to get a difference, libstatgrab returned NaN
    # Allow it initially to get _stats_pct set but not after
    return if $self->_stats_pct && $pcts->idle eq 'NaN';

    $self->_stats_pct($pcts);
    $self->_last_stats($stats);
    $self->_set_last_update(time);
}

#
# (Possibly) adjust our max_procs before the call to start().
#
before start => sub {
    my $self = shift;

    return if time - $self->last_update < $self->update_frequency;

    $self->update_stats_pct;

    my $new_procs;
    my $min_ok = max(  0, $self->idle_target - $self->idle_threshold);
    my $max_ok = min(100, $self->idle_target + $self->idle_threshold);

    #
    # It's possible for idle to be NaN if not enough time has elapsed between
    # the initial call to update_stats_pct and the latest call. In this case
    # neither check against $self->idle will be true and no update will occur
    #
    if ($self->idle >= $max_ok && $self->running_procs >= $self->max_procs) {
        $new_procs = $self->adjust_up;

    } elsif ($self->idle <= $min_ok) {
        $new_procs = $self->adjust_down;
    }

    my $prev_procs = $self->max_procs;

    $self->set_max_procs($new_procs)
        if $new_procs;

    $self->run_on_update->($self, $prev_procs)
        if ($self->run_on_update && ref($self->run_on_update) eq 'CODE');
};

#
# constrain max_procs to be within our soft min and max
#
around set_max_procs => sub {
    my ($orig, $self, $new_val) = @_;

    $orig->($self,
        min( $self->soft_max_procs, max($self->soft_min_procs, $new_val)
        )
    );
};

sub stats {
    my $self = shift;
    my $prev_procs = shift // $self->max_procs;

    sprintf(
        "%5.1f id %3d run %3d omax %3d nmax %3d smin %3d smax %3d hmin %3d hmax",
        $self->idle,
        scalar($self->running_procs),
        $prev_procs,
        $self->max_procs,
        $self->soft_min_procs,
        $self->soft_max_procs,
        $self->hard_min_procs,
        $self->hard_max_procs
    );
}

sub dump_stats {
    my $self = shift;
    print STDERR $self->stats(@_)."\n";
}

#
# Increase soft_max_procs to a maximum of hard_max_procs
#
# We'll use the system's idle percentage to tell us how much
# to increase by, the more idle the system is, the more we'll
# allow soft_max_procs to grow. Hopefully this will allow us
# to quickly adjust to the system without over-loading it if
# it's already close to our target idle state
#
sub adjust_soft_max {
    my $self = shift;
    $self->soft_max_procs(
        min($self->hard_max_procs,
            $self->soft_max_procs
            + max(1, int(
                ($self->hard_max_procs - $self->max_procs)
                * ($self->idle - $self->idle_target)
                / 100
            ))
        )
    );
}

#
# Decrease soft_min_procs, the system is too busy
#
sub adjust_soft_min {
    my $self = shift;
    $self->soft_min_procs(
        max($self->hard_min_procs,
            $self->hard_min_procs
            + max(0, int(
                ($self->max_procs - $self->hard_min_procs)
                * ($self->idle_target - $self->idle)
                / 100
            ))
        )
    );
}

#
# Adjust our number of running processes (max_procs) to half way between
# the current number and our soft max. If we're already at
# soft max, try to adjust the soft max up first.
#
# Set the soft min to the current number of running procs
# as it wasn't enough to hit our idle target so we shouldn't
# go below it again (although we can if we actually need to).
#
sub adjust_up {
    my $self = shift;
    my $cur = $self->max_procs;

    my $max = $cur >= $self->soft_max_procs
        ? $self->adjust_soft_max
        : $self->soft_max_procs;

    $self->soft_min_procs($cur);
    $cur + max(1,int(($max - $cur)/2));
}

sub adjust_down {
    my $self = shift;
    my $cur = $self->max_procs;

    my $min = $cur <= $self->soft_min_procs
        ? $self->adjust_soft_min
        : $self->soft_min_procs;

    # Shouldn't happen, but test for it anyway
    return undef unless $cur > $min;

    $self->soft_max_procs($cur);
    $min + int(($cur - $min)/2);
}


#
# libstatgrab doesn't like freeze/thaw (saw assertion errors from vector.c)
# so we need to set those # attributes that house Unix::Statgrab objects to
# undef before # being frozen. Restore them after freezing.
#
# Also, freeze/thaw can't handle CODE references so we'll clear
# our run_on_update hook. There will still be problems with the
# underlying Parallel::ForkManager hooks but I'm not going to
# try to fix those here. That should be handled by Parallel::ForkManager
# I believe.
#
sub STORABLE_freeze {
    my ($self, $cloning) = @_;
    state $storing = 0;
    return if $cloning || $storing;

    # libstatgrab isn't happy when it's frozen / thawed
    my %save;
    for (@{$self->__unstorable}) {
        $save{$_} = $self->$_;
        $self->$_(undef);
    }
    $save{run_on_update} = $self->run_on_update;
    $self->clear_run_on_update;

    $storing = 1;
    my $ret = freeze($self);
    $storing = 0;

    $self->$_($save{$_}) for @{$self->__unstorable};
    $self->run_on_update($save{run_on_update});

    $ret;
};

#
# Since our Unix::Statgrab objects are all lazily built, they were
# set to undef before freeze(). We need to clear them in thaw() so
# they can be re-built. Not perfect but should keep things working
#
# We will have lost the run_on_update hook if it was set, but nothing
# to be done about that.
#
sub STORABLE_thaw {
    my ($self, $cloning, $data) = @_;
    state $thawing = 0;

    return if $cloning || $thawing;

    $thawing = 1;
    %$self = %{thaw($data)};

    eval "\$self->_clear$_" for @{$self->__unstorable};

    # And this non-hidden code ref
    $self->clear_run_on_update;

    $thawing = 0;
}

1;

__END__