harness

#!/opt/maths/bin/perl
use strict;
use warnings;
use lib ('../lib', 'lib');
use Seq::Db;
use H::Heap;
use Math::GMP;
use POSIX qw{ :sys_wait_h };
use Type;

print "kill -HUP $$\n";

my($opt_recreate, $debug, $opt_strategy, $opt_gain) = (0, 0, 0, 0);
my($maxproc, $maxrun, $maxload, $limit, $drain) = (10, 100, 0, 0);
my $typename = 't';
my %override;
$SIG{'HUP'} = sub { $drain = 1 };
while (@ARGV && $ARGV[0] =~ /^-/) {
    my $arg = shift(@ARGV);
    last if $arg eq '--';
    $opt_recreate = 1, next if $arg eq '-r';
    $opt_strategy = 1, next if $arg eq '-s';
    ($maxload = $1 || shift(@ARGV)), next if $arg =~ /^-a(\d*)$/;
    ($limit = 1, $maxload = $1 || shift(@ARGV)), next if $arg =~ /^-A(\d*)$/;
    $debug = 1, next if $arg eq '-d';
    $override{$1} = $2, next if $arg =~ /^-p(-?\d+)=(-?\d+(?:\.\d+)?)$/;
    $opt_gain = $1, next if $arg =~ /^-g(-?\d+(?:\.\d+)?)$/;
    $typename = $1, next if $arg =~ /^-y(.*)$/;
    die "Unknown option '$arg'";
}
$maxproc = shift(@ARGV) if @ARGV;
$maxrun = shift(@ARGV) if @ARGV;
$maxload ||= $maxproc;

my $type = Type->new($typename);
$type->bind_owner('harness');

local @ENV{qw{ DBI_TRACE DBIC_TRACE }} = (1, 1) if $debug;
my $db = Seq::Db->new($type, $opt_recreate);

use Data::Dumper;
local $Data::Dumper::Indent = $Data::Dumper::Sortkeys = 1;

Seq::Run->restrategise($db) if $opt_strategy;
my $prio = H::Heap->new(q{ ::prio($b) <=> ::prio($a) });
$prio->insert(Seq::Starter->new($type, 0, $maxload));
my(%pid, @runnable);

while (1) {
    if ($drain) {
        printf "Draining %s processes\n", 0 + keys %pid;
        waitFor($db, \%pid, 1) while keys %pid;
        last;
    }

    if (@runnable) {
        if (keys(%pid) >= $maxproc) {
            waitFor($db, \%pid, 1);
        } else {
            my $r = shift @runnable;
            $override{$r->n} -= $opt_gain;
            my $pid = $r->run($db);
            if ($pid) {
                $pid{$pid} = $r;
                $drain = 1, next if --$maxrun <= 0;
            }
        }
    } elsif (my $obj = $prio->fetch) {
        $prio->insert($_) for $obj->prep($db);
        push @runnable, $obj->runnable;
    } else {
        waitFor($db, \%pid, 0);
    }
}
exit 0;

sub prio {
    my($obj) = @_;
    return $obj->rprio($type) + ($override{$obj->n} // 0);
}

sub waitFor {
    my($db, $pids, $block) = @_;
    my @t0 = times;
    my $pid = waitpid(-1, $block ? 0 : WNOHANG);
    return 0 if $pid <= 0;

    my $obj = delete $pids->{$pid};
    my @t1 = times;
    my $time = $t1[2] + $t1[3] - $t0[2] - $t0[3];
    $prio->insert($_) for $obj->finalize($db, $time);
    return 1;
}

package Seq::Starter {
    use List::Util qw{ max };
    sub new {
        my($class, $type, $start, $count) = @_;
        my $self = bless {
            type => $type,
            start => $start,
            count => $count,
        }, $class;
        warn sprintf "new starter: from %s for %s prio %.2f\n",
                $start, $count, $self->rprio($type);
        return $self;
    }
    sub n { -$_[0]->{start} }
    sub type { $_[0]->{type} }
    sub rprio {
        my($self, $type) = @_;
        my $start = $self->{start};
        my $count = $self->{count};
        # We don't want to block fetching of later entries just because
        # there are low priority ones in the way.
        return $self->{rprio} //= max(
            map $type->gprio($_), grep $_ > 1, $start .. $start + $count
        );
    }
    sub prep {
        my($self, $db) = @_;
        my $start = $self->{start};
        my $count = $self->{count};
        my $end = $start + $count - 1;
        my $max = Seq::TauG->max_known($db);
        if ($end > $max) {
            Seq::TauG->genTo($db, $end);
        }
        return (
            ($limit ? () : Seq::Starter->new($self->type, $end + 1, $count)),
            Seq::TauG->range($db, $start, $end),
        );
    }
    sub runnable { () }
};