hydra/src/script/hydra-queue-runner

#! /var/run/current-system/sw/bin/perl -w

use strict;
use Cwd;
use File::Basename;
use POSIX qw(dup2 :sys_wait_h);
use Hydra::Schema;
use Hydra::Helper::Nix;
use Hydra::Model::DB;
use IO::Handle;
use Nix::Store;

chdir Hydra::Model::DB::getHydraPath or die;
my $db = Hydra::Model::DB->new();

STDOUT->autoflush();

my $lastTime;

#$SIG{CHLD} = 'IGNORE';


sub unlockDeadBuilds {
    # Unlock builds whose building process has died.
    txn_do($db, sub {
        my @builds = $db->resultset('Builds')->search({finished => 0, busy => 1});
        foreach my $build (@builds) {
            my $pid = $build->locker;
            my $unlock = 0;
            if ($pid == $$) {
                if (!defined $lastTime || $build->starttime < $lastTime - 300) {
                    $unlock = 1;
                }
            } elsif (kill(0, $pid) != 1) { # see if we can signal the process
                $unlock = 1;
            }
            if ($unlock) {
                print "build ", $build->id, " pid $pid died, unlocking\n";
                $build->update({ busy => 0, locker => ""});
                $build->buildsteps->search({ busy => 1 })->update({ busy => 0, status => 4, stoptime => time });
            }
        }
    });
}


# Given a build, return an arbitrary queued build on which this build
# depends; or undef if no such build exists.
sub findBuildDependencyInQueue {
    my ($buildsByDrv, $build) = @_;
    return undef unless isValidPath($build->drvpath);
    my @deps = grep { /\.drv$/ && $_ ne $build->drvpath } computeFSClosure(0, 0, $build->drvpath);
    return unless scalar @deps > 0;
    foreach my $d (@deps) {
        my $b = $buildsByDrv->{$d};
        next unless defined $b;
        return $db->resultset('Builds')->find($b);
    }
    return undef;
}


sub checkBuilds {
    # print "looking for runnable builds...\n";

    my @buildsStarted;

    my $machines = getMachines;

    my %maxConcurrent;

    foreach my $machineName (keys %{$machines}) {
        foreach my $system (${$machines}{$machineName}{'systemTypes'}) {
            $maxConcurrent{$system} = (${$machines}{$machineName}{'maxJobs'} or 0) + ($maxConcurrent{$system} or 0)
        }
    }

    txn_do($db, sub {

        # Cache scheduled by derivation path to speed up
        # findBuildDependencyInQueue.
        my $buildsByDrv = {};
        $buildsByDrv->{$_->drvpath} = $_->id
            foreach $db->resultset('Builds')->search({ finished => 0, enabled => 1 }, { join => ['project'] });

        # Get the system types for the runnable builds.
        my @systemTypes = $db->resultset('Builds')->search(
            { finished => 0, busy => 0, enabled => 1 },
            { join => ['project'], select => ['system'], as => ['system'], distinct => 1 });

        # For each system type, select up to the maximum number of
        # concurrent build for that system type.  Choose the highest
        # priority builds first, then the oldest builds.
        foreach my $system (@systemTypes) {
            # How many builds are already currently executing for this
            # system type?
            my $nrActive = $db->resultset('Builds')->search(
                {finished => 0, busy => 1, system => $system->system})->count;

            (my $systemTypeInfo) = $db->resultset('SystemTypes')->search({system => $system->system});
            my $max = defined $systemTypeInfo ? $systemTypeInfo->maxconcurrent : $maxConcurrent{$system->system} // 2;

            my $extraAllowed = $max - $nrActive;
            $extraAllowed = 0 if $extraAllowed < 0;

            # Select the highest-priority builds to start.
            my @builds = $extraAllowed == 0 ? () : $db->resultset('Builds')->search(
                { finished => 0, busy => 0, system => $system->system, enabled => 1 },
                { join => ['project'], order_by => ["priority DESC", "id"] });

            my $started = 0;
            foreach my $build (@builds) {
                # Find a dependency of $build that has no queued
                # dependencies itself.  This isn't strictly necessary,
                # but it ensures that Nix builds are done as part of
                # their corresponding Hydra builds, rather than as a
                # dependency of some other Hydra build.
                while (my $dep = findBuildDependencyInQueue($buildsByDrv, $build)) {
                    $build = $dep;
                }
                next if $build->busy;

                my $logfile = getcwd . "/logs/" . $build->id;
                mkdir(dirname $logfile);
                unlink($logfile);
                $build->update(
                    { busy => 1
                    , locker => $$
                    , logfile => $logfile
                    , starttime => time()
                    });
                push @buildsStarted, $build;

                last if ++$started >= $extraAllowed;
            }

            if ($started > 0) {
                print STDERR "system type `", $system->system,
                    "': $nrActive active, $max allowed, started $started builds\n";
            }
        }
    });

    # Actually start the builds we just selected.  We need to do this
    # outside the transaction in case it aborts or something.
    foreach my $build (@buildsStarted) {
        my $id = $build->id;
        print "starting build $id (", $build->project->name, ":", $build->jobset->name, ':', $build->job->name, ") on ", $build->system, "\n";
        eval {
            my $logfile = $build->logfile;
            my $child = fork();
            die unless defined $child;
            if ($child == 0) {
                eval {
                    open LOG, ">$logfile" or die "cannot create logfile $logfile";
                    POSIX::dup2(fileno(LOG), 1) or die;
                    POSIX::dup2(fileno(LOG), 2) or die;
                    exec("hydra-build", $id);
                };
                warn "cannot start build $id: $@";
                POSIX::_exit(1);
            }
        };
        if ($@) {
            warn $@;
            txn_do($db, sub {
                $build->busy(0);
                $build->locker($$);
                $build->update;
            });
        }
    }
}


if (scalar(@ARGV) == 1 && $ARGV[0] eq "--unlock") {
    unlockDeadBuilds;
    exit 0;
}


while (1) {
    eval {
        # Clean up zombies.
        while ((waitpid(-1, &WNOHANG)) > 0) { };

        unlockDeadBuilds;

        checkBuilds;

        $lastTime = time();
    };
    warn $@ if $@;

    # print "sleeping...\n";
    sleep(5);
}