grahamc/hydra - Change 4YCF3KBGI4VYKHJXAREJLCJLY3UWB2FX447CJ4XQWFRKRFKG5WCQC

Concurrent hydra-evaluator

This rewrites the top-level loop of hydra-evaluator in C++. The Perl stuff is moved into hydra-eval-jobset. (Rewriting the entire evaluator would be nice but is a bit too much work.) The new version has some advantages:

It can run multiple jobset evaluations in parallel.
It uses PostgreSQL notifications so it doesn't have to poll the database. So if a jobset is triggered via the web interface or from a GitHub / Bitbucket webhook, evaluation of the jobset will start almost instantaneously (assuming the evaluator is not at its concurrency limit).
It imposes a timeout on evaluations. So if e.g. hydra-eval-jobset hangs connecting to a Mercurial server, it will eventually be killed.

Created by Eelco Dolstra on October 13, 2016

4YCF3KBGI4VYKHJXAREJLCJLY3UWB2FX447CJ4XQWFRKRFKG5WCQC

Dependencies

In channels

main

upstream

Change contents

Replacement in configure.ac at line 14 [12.15951]
B:BD[11.1] → [7.0:24]
```
CXXFLAGS+=" -std=c++11"
```
[11.1]
[12.16096]
```
CXXFLAGS+=" -std=c++17"
```
Insertion in configure.ac at line 73 [12.15951]
[12.16466]
[13.38]
```
  src/hydra-evaluator/Makefile
```

Replacement in release.nix at line 61 [15.1]

B:BD[14.565] → [9.0:135]

          rev = "edf9eb8181e01f6b2123e5690019cfeeb44fc1c2";
          sha256 = "1a00q9pypfziyi9hxl4rsammhwj7991wm4b1z9zcgl7zqksr3582";

[14.565]

[14.700]

          rev = "5e61b422c58baac26b232233d39f5814cc35d52a";
          sha256 = "0awic5zwibgpj5shpgjf2364imp2f84c8xi5r0x4p351q4kpg9z4";

Insertion in release.nix at line 127 [15.1]
[15.926]
[16.1760]
```
      stdenv = overrideCC stdenv gcc6;
```

Replacement in src/Makefile.am at line 1 [17.21]

B:BD[17.21] → [18.112:185]

SUBDIRS = hydra-eval-jobs hydra-queue-runner sql script lib root xsl ttf

[17.21]

[17.62]

SUBDIRS = hydra-evaluator hydra-eval-jobs hydra-queue-runner sql script lib root xsl ttf

File addition: hydra-evaluator (d--x------)
[19.4]

File addition: Makefile.am (----------)

[0.325]

bin_PROGRAMS = hydra-evaluator
hydra_evaluator_SOURCES = hydra-evaluator.cc
hydra_evaluator_LDADD = $(NIX_LIBS) -lpqxx
hydra_evaluator_CXXFLAGS = $(NIX_CFLAGS) -Wall -I ../libhydra

File addition: hydra-evaluator.cc (----------)

[0.325]

#include "shared.hh"
#include "db.hh"
#include "pool.hh"
#include <algorithm>
#include <thread>
#include <cstring>
#include <sys/types.h>
#include <sys/wait.h>
using namespace nix;
struct Evaluator
{
    nix::Pool<Connection> dbPool;
    typedef std::pair<std::string, std::string> JobsetName;
    struct Jobset
    {
        JobsetName name;
        time_t lastCheckedTime, triggerTime;
        int checkInterval;
        Pid pid;
    };
    typedef std::map<JobsetName, Jobset> Jobsets;
    int evalTimeout = 3600;
    size_t maxEvals = 4;
    struct State
    {
        size_t runningEvals = 0;
        Jobsets jobsets;
    };
    Sync<State> state_;
    std::condition_variable childStarted;
    std::condition_variable maybeDoWork;
    const time_t notTriggered = std::numeric_limits<time_t>::max();
    void readJobsets()
    {
        auto conn(dbPool.get());
        pqxx::work txn(*conn);
        auto res = txn.parameterized
            ("select project, j.name, lastCheckedTime, triggerTime, checkInterval from Jobsets j join Projects p on j.project = p.name "
             "where j.enabled != 0 and p.enabled != 0").exec();
        auto state(state_.lock());
        std::set<JobsetName> seen;
        for (auto const & row : res) {
            auto name = JobsetName{row["project"].as<std::string>(), row["name"].as<std::string>()};
            auto res = state->jobsets.try_emplace(name, Jobset{name});
            auto & jobset = res.first->second;
            jobset.lastCheckedTime = row["lastCheckedTime"].as<time_t>(0);
            jobset.triggerTime = row["triggerTime"].as<time_t>(notTriggered);
            jobset.checkInterval = row["checkInterval"].as<time_t>();
            seen.insert(name);
        }
        for (auto i = state->jobsets.begin(); i != state->jobsets.end(); )
            if (seen.count(i->first))
                ++i;
            else {
                printInfo("forgetting jobset ‘%s:%s’", i->first.first, i->first.second);
                i = state->jobsets.erase(i);
            }
    }
    void startEval(State & state, Jobset & jobset)
    {
        printInfo("starting evaluation of jobset ‘%s:%s’", jobset.name.first, jobset.name.second);
        assert(jobset.pid == -1);
        jobset.pid = startProcess([&]() {
            Strings args = { "timeout", "-s", "KILL", std::to_string(evalTimeout), "hydra-eval-jobset", jobset.name.first, jobset.name.second };
            execvp(args.front().c_str(), stringsToCharPtrs(args).data());
            throw SysError(format("executing ‘%1%’") % args.front());
        });
        state.runningEvals++;
        childStarted.notify_one();
        time_t now = time(0);
        {
            auto conn(dbPool.get());
            pqxx::work txn(*conn);
            txn.parameterized
                ("update Jobsets set lastCheckedTime = $1, triggerTime = null where project = $2 and name = $3")
                (now)
                (jobset.name.first)
                (jobset.name.second)
                .exec();
            txn.commit();
            jobset.lastCheckedTime = now;
            jobset.triggerTime = notTriggered;
        }
    }
    void startEvals(State & state)
    {
        std::vector<Jobsets::iterator> sorted;
        time_t now = time(0);
        /* Filter out jobsets that have been evaluated recently and have
           not been triggered. */
        for (auto i = state.jobsets.begin(); i != state.jobsets.end(); ++i)
            if (i->second.pid == -1 &&
                (i->second.triggerTime != std::numeric_limits<time_t>::max() ||
                    (i->second.checkInterval > 0 && i->second.lastCheckedTime + i->second.checkInterval <= now)))
                sorted.push_back(i);
        /* Put jobsets in order of ascending trigger time, last checked
           time, and name. */
        std::sort(sorted.begin(), sorted.end(),
            [](const Jobsets::iterator & a, const Jobsets::iterator & b) {
                return
                    a->second.triggerTime != b->second.triggerTime
                    ? a->second.triggerTime < b->second.triggerTime
                    : a->second.lastCheckedTime != b->second.lastCheckedTime
                      ? a->second.lastCheckedTime < b->second.lastCheckedTime
                      : a->first < b->first;
            });
        /* Start jobset evaluations up to the concurrency limit.*/
        for (auto & i : sorted) {
            if (state.runningEvals >= maxEvals) break;
            startEval(state, i->second);
        }
    }
    void loop()
    {
        auto state(state_.lock());
        while (true) {
            time_t now = time(0);
            std::chrono::seconds sleepTime = std::chrono::seconds::max();
            if (state->runningEvals < maxEvals) {
                for (auto & i : state->jobsets)
                    if (i.second.pid == -1 &&
                        i.second.checkInterval > 0)
                        sleepTime = std::min(sleepTime, std::chrono::seconds(
                                std::max((time_t) 1, i.second.lastCheckedTime - now + i.second.checkInterval)));
            }
            debug("waiting for %d s", sleepTime.count());
            if (sleepTime == std::chrono::seconds::max())
                state.wait(maybeDoWork);
            else
                state.wait_for(maybeDoWork, sleepTime);
            startEvals(*state);
        }
    }
    /* A thread that listens to PostgreSQL notifications about jobset
       changes, updates the jobsets map, and signals the main thread
       to start evaluations. */
    void databaseMonitor()
    {
        while (true) {
            try {
                auto conn(dbPool.get());
                receiver jobsetsAdded(*conn, "jobsets_added");
                receiver jobsetsDeleted(*conn, "jobsets_deleted");
                receiver jobsetsChanged(*conn, "jobset_scheduling_changed");
                while (true) {
                    /* Note: we read/notify before
                       await_notification() to ensure we don't miss a
                       state change. */
                    readJobsets();
                    maybeDoWork.notify_one();
                    conn->await_notification();
                    printInfo("received jobset event");
                }
            } catch (std::exception & e) {
                printError("exception in database monitor thread: %s", e.what());
                sleep(30);
            }
        }
    }
    /* A thread that reaps child processes.*/
    void reaper()
    {
        while (true) {
            {
                auto state(state_.lock());
                while (!state->runningEvals)
                    state.wait(childStarted);
            }
            int status;
            pid_t pid = waitpid(-1, &status, 0);
            if (pid == -1) {
                if (errno == EINTR) continue;
                throw SysError("waiting for children");
            }
            {
                auto state(state_.lock());
                assert(state->runningEvals);
                state->runningEvals--;
                for (auto & jobset : state->jobsets)
                    if (jobset.second.pid == pid) {
                        printInfo("evaluation of jobset ‘%s:%s’ finished with status %d",
                            jobset.first.first, jobset.first.second, status);
                        jobset.second.pid.release();
                        maybeDoWork.notify_one();
                        break;
                    }
            }
        }
    }
    void run()
    {
        std::thread reaperThread([&]() { reaper(); });
        std::thread monitorThread([&]() { databaseMonitor(); });
        while (true) {
            try {
                loop();
            } catch (std::exception & e) {
                printError("exception in main loop: %s", e.what());
                sleep(30);
            }
        }
    }
};
int main(int argc, char * * argv)
{
    return handleExceptions(argv[0], [&]() {
        initNix();
        signal(SIGINT, SIG_DFL);
        signal(SIGTERM, SIG_DFL);
        signal(SIGHUP, SIG_DFL);
        parseCmdLine(argc, argv, [&](Strings::iterator & arg, const Strings::iterator & end) {
            return false;
        });
        Evaluator().run();
    });
}

Replacement in src/hydra-queue-runner/Makefile.am at line 7 [18.219]
B:BD[18.368] → [18.368:369]
B:BD[18.369] → [10.0:34]
```
AM_CXXFLAGS = $(NIX_CFLAGS) -Wall
```
[18.368]
```
hydra_queue_runner_CXXFLAGS = $(NIX_CFLAGS) -Wall -I ../libhydra
```
File addition: libhydra (d--x------)
[19.4]
File move: db.hh → db.hh
BF:BFD[18.187] → [20.1022:1051]
BF:BF[20.1051] → [20.93:93]
[0.9041]
[20.93]
Replacement in src/script/Makefile.am at line 7 [12.2198648]
B:BD[21.442] → [21.442:465]
```
  hydra-evaluator				\
```
[21.422]
[21.491]
```
  hydra-eval-jobset				\
```
File move: hydra-evaluator → hydra-eval-jobset
BF:BFD[22.2543] → [23.403:442]
BF:BF[23.442] → [24.1319:1319]
[22.2543]
[24.1319]

Replacement in src/script/hydra-eval-jobset at line 348 [24.1319]

B:BD[25.477] → [25.477:499]

B:BD[25.499] → [26.2600:2660]

∅:D[26.2660] → [25.499:553]

B:BD[25.499] → [25.499:553]

B:BD[25.553] → [4.0:46]

∅:D[4.46] → [26.2857:2931]

B:BD[26.2857] → [26.2857:2931]

sub checkSomeJobset {
    # If any jobset has been triggered by a push, check it.
    my ($jobset) = $db->resultset('Jobsets')->search(
        { 'triggertime' => { '!=', undef } },
        { join => 'project', order_by => [ 'triggertime' ], rows => 1 });

[27.9901]

[26.2931]

die "syntax: $0 <PROJECT> <JOBSET>\n" unless @ARGV == 2;

Replacement in src/script/hydra-eval-jobset at line 350 [24.1319]

B:BD[26.2932] → [26.2932:2999]

B:BD[26.2999] → [28.1015:1111]

∅:D[28.1111] → [26.3086:3137]

B:BD[26.3086] → [26.3086:3137]

B:BD[26.3137] → [29.1609:1674]

∅:D[29.1674] → [28.1112:1270]

B:BD[26.3190] → [28.1112:1270]

∅:D[28.1270] → [26.3191:3312]

B:BD[30.308] → [26.3191:3312]

∅:D[30.398] → [25.744:783]

∅:D[26.3312] → [25.744:783]

B:BD[25.744] → [25.744:783]

B:BD[25.783] → [3.0:67]

∅:D[3.67] → [31.0:4]

∅:D[25.824] → [31.0:4]

B:BD[27.10180] → [31.0:4]

B:BD[31.56] → [31.56:144]

B:BD[31.144] → [8.0:160]

∅:D[32.124] → [33.64:95]

∅:D[8.160] → [33.64:95]

B:BD[31.228] → [33.64:95]

∅:D[33.95] → [27.10180:10184]

∅:D[31.284] → [27.10180:10184]

B:BD[27.10180] → [27.10180:10184]

B:BD[27.10184] → [34.968:980]

B:BD[34.980] → [35.0:11]

B:BD[35.11] → [25.852:1029]

B:BD[25.1029] → [2.0:44]

∅:D[2.44] → [25.1071:1103]

B:BD[25.1071] → [25.1071:1103]

∅:D[25.1103] → [35.30:37]

∅:D[36.1461] → [35.30:37]

B:BD[35.30] → [35.30:37]

B:BD[35.37] → [37.3067:3102]

∅:D[38.869] → [34.1036:1038]

B:BD[34.1036] → [34.1036:1038]

    # Otherwise, check the jobset that hasn't been checked for the
    # longest time (but don't check more often than the jobset's
    # minimal check interval).
    ($jobset) = $db->resultset('Jobsets')->search(
        { 'project.enabled' => 1, 'me.enabled' => { '!=' => 0 },
        , 'checkinterval' => { '!=', 0 }
        , -or => [ 'lastcheckedtime' => undef, 'lastcheckedtime' => { '<', \ (time() . " - me.checkinterval") } ] },
        { join => 'project', order_by => [ 'lastcheckedtime nulls first' ], rows => 1 })
        unless defined $jobset;
    return 0 unless defined $jobset;
    return system($0, $jobset->project->name, $jobset->name) == 0;
}
if (scalar @ARGV == 2) {
    my $projectName = $ARGV[0];
    my $jobsetName = $ARGV[1];
    my $jobset = $db->resultset('Jobsets')->find($projectName, $jobsetName) or 
      die "$0: specified jobset \"$projectName:$jobsetName\" does not exist\n";
    exit checkJobset($jobset);
}
while (1) {
    eval {
        if (checkSomeJobset) {
            # Just so we don't go completely crazy if lastcheckedtime
            # isn't updated properly.
            sleep 1;
        } else {
            # print STDERR "sleeping...\n";
            sleep 30;
        }
    };
    if ($@) { print STDERR "$@"; }
}

[26.2932]

my $projectName = $ARGV[0];
my $jobsetName = $ARGV[1];
my $jobset = $db->resultset('Jobsets')->find($projectName, $jobsetName) or
    die "$0: specified jobset \"$projectName:$jobsetName\" does not exist\n";
exit checkJobset($jobset);

Insertion in src/sql/hydra.sql at line 85 [24.1]

[5.1688]


create function notifyJobsetsAdded() returns trigger as 'begin notify jobsets_added; return null; end;' language plpgsql;
create trigger JobsetsAdded after insert on Jobsets execute procedure notifyJobsetsAdded();

Insertion in src/sql/hydra.sql at line 89 [24.1]

[5.1689]

create function notifyJobsetsDeleted() returns trigger as 'begin notify jobsets_deleted; return null; end;' language plpgsql;
create trigger JobsetsDeleted after delete on Jobsets execute procedure notifyJobsetsDeleted();
create function notifyJobsetSchedulingChanged() returns trigger as 'begin notify jobset_scheduling_changed; return null; end;' language plpgsql;
create trigger JobsetSchedulingChanged after update on Jobsets for each row
  when (((old.triggerTime is distinct from new.triggerTime) and (new.triggerTime is not null))
        or old.checkInterval != new.checkInterval
        or old.enabled != new.enabled)
  execute procedure notifyJobsetSchedulingChanged();

File addition: upgrade-50.sql (----------)

[22.3004]

create function notifyJobsetsAdded() returns trigger as 'begin notify jobsets_added; return null; end;' language plpgsql;
create trigger JobsetsAdded after insert on Jobsets execute procedure notifyJobsetsAdded();
create function notifyJobsetsDeleted() returns trigger as 'begin notify jobsets_deleted; return null; end;' language plpgsql;
create trigger JobsetsDeleted after delete on Jobsets execute procedure notifyJobsetsDeleted();
create function notifyJobsetSchedulingChanged() returns trigger as 'begin notify jobset_scheduling_changed; return null; end;' language plpgsql;
create trigger JobsetSchedulingChanged after update on Jobsets for each row
  when ((old.triggerTime is distinct from new.triggerTime) and (new.triggerTime is not null))
        or old.checkInterval != new.checkInterval
        or old.enabled != new.enabled
  execute procedure notifyJobsetSchedulingChanged();

Replacement in tests/Makefile.am at line 16 [41.92]

B:BD[39.58] → [40.106:230]

  PATH=$(abs_top_srcdir)/src/script:$(abs_top_srcdir)/src/hydra-eval-jobs:$(abs_top_srcdir)/src/hydra-queue-runner:$$PATH \

[39.58]

[42.37753]

  PATH=$(abs_top_srcdir)/src/hydra-evaluator:$(abs_top_srcdir)/src/script:$(abs_top_srcdir)/src/hydra-eval-jobs:$(abs_top_srcdir)/src/hydra-queue-runner:$$PATH \

Replacement in tests/Setup.pm at line 64 [45.518]

B:BD[43.1336] → [44.9344:9463]

    my ($res, $stdout, $stderr) = captureStdoutStderr(60, ("hydra-evaluator", $jobset->project->name, $jobset->name));

[43.1336]

[43.1469]

    my ($res, $stdout, $stderr) = captureStdoutStderr(60, ("hydra-eval-jobset", $jobset->project->name, $jobset->name));

Replacement in tests/api-test.pl at line 52 [47.36450]
B:BD[46.245] → [6.175:217]
```
system("hydra-evaluator sample default");
```
[46.245]
[47.38737]
```
system("hydra-eval-jobset sample default");
```

Replacement in tests/api-test.pl at line 59 [47.36450]

B:BD[47.39017] → [6.218:293]

system("echo >> /run/jobset/default.nix; hydra-evaluator sample default");

[47.39017]

[47.39299]

system("echo >> /run/jobset/default.nix; hydra-eval-jobset sample default");

Replacement in tests/s3-backup-test.pl at line 22 [44.9772]

B:BD[44.10569] → [44.10569:10644]

system("hydra-evaluator " . $jobset->project->name . " " . $jobset->name);

[44.10569]

[44.10644]

system("hydra-eval-jobset " . $jobset->project->name . " " . $jobset->name);

Concurrent hydra-evaluator

Dependencies

In channels

Change contents

Replacement in configure.ac at line 14 [12.15951]

Insertion in configure.ac at line 73 [12.15951]

Replacement in release.nix at line 61 [15.1]

Insertion in release.nix at line 127 [15.1]

Replacement in src/Makefile.am at line 1 [17.21]

File addition: hydra-evaluator (d--x------)

File addition: Makefile.am (----------)

File addition: hydra-evaluator.cc (----------)

Replacement in src/hydra-queue-runner/Makefile.am at line 7 [18.219]

File addition: libhydra (d--x------)

File move: db.hh → db.hh

Replacement in src/script/Makefile.am at line 7 [12.2198648]

File move: hydra-evaluator → hydra-eval-jobset