lix-project/hydra
3
0
Fork 6
Code Issues 2 Pull requests Packages Projects Releases Wiki Activity

Compare commits

base: lix-project:main
Branches Tags
lix-project:main
lix-project:lix-update
snaakey:main
..
compare: snaakey:main
Branches Tags
snaakey:main
lix-project:main
lix-project:lix-update

No commits in common. "main" and "main" have entirely different histories.
main ... main

19 changed files with 197 additions and 268 deletions
Show stats Expand all files Collapse all files

12
.clang-tidy
View file

@ -1,12 +0,0 @@
UseColor: true
Checks:
- -*
- bugprone-*
# kind of nonsense
- -bugprone-easily-swappable-parameters
# many warnings due to not recognizing `assert` properly
- -bugprone-unchecked-optional-access
- modernize-*
- -modernize-use-trailing-return-type

2
.envrc
View file

@ -1 +1 @@
use flake .#clang use flake

52
flake.lock
View file

@ -24,11 +24,11 @@
] ]
}, },
"locked": { "locked": {
"lastModified": 1730504689, "lastModified": 1722555600,
"narHash": "sha256-hgmguH29K2fvs9szpq2r3pz2/8cJd2LPS+b4tfNFCwE=", "narHash": "sha256-XOQkdLafnb/p9ij77byFQjDf5m5QYl9b2REiVClC+x4=",
"owner": "hercules-ci", "owner": "hercules-ci",
"repo": "flake-parts", "repo": "flake-parts",
"rev": "506278e768c2a08bec68eb62932193e341f55c90", "rev": "8471fe90ad337a8074e957b69ca4d0089218391d",
"type": "github" "type": "github"
}, },
"original": { "original": {
@ -48,11 +48,11 @@
"pre-commit-hooks": "pre-commit-hooks" "pre-commit-hooks": "pre-commit-hooks"
}, },
"locked": { "locked": {
"lastModified": 1731683711, "lastModified": 1723919517,
"narHash": "sha256-bq21I1EjXJa/s5Rra9J9ot2NkPCnI0F5uNPurwYLdpE=", "narHash": "sha256-D6+zmRXzr85p7riphuIrJQqangoJe70XM5jHhMWwXws=",
"ref": "refs/heads/main", "ref": "refs/heads/main",
"rev": "c859d03013712b349d82ee6223948d6d03e63a8d", "rev": "278fddc317cf0cf4d3602d0ec0f24d1dd281fadb",
"revCount": 16489, "revCount": 16138,
"type": "git", "type": "git",
"url": "https://git.lix.systems/lix-project/lix" "url": "https://git.lix.systems/lix-project/lix"
}, },
@ -74,11 +74,11 @@
"treefmt-nix": "treefmt-nix" "treefmt-nix": "treefmt-nix"
}, },
"locked": { "locked": {
"lastModified": 1731890968, "lastModified": 1723579251,
"narHash": "sha256-6xMxT2duVMO6fo1AXfTjqh7LW3ZmNiHw6kBaAhweLGo=", "narHash": "sha256-xnHtfw0gRhV+2S9U7hQwvp2klTy1Iv7FlMMO0/WiMVc=",
"ref": "refs/heads/main", "ref": "refs/heads/main",
"rev": "912a9d63319e71ca131e16eea3348145a255db2e", "rev": "42a160bce2fd9ffebc3809746bc80cc7208f9b08",
"revCount": 616, "revCount": 609,
"type": "git", "type": "git",
"url": "https://git.lix.systems/lix-project/nix-eval-jobs" "url": "https://git.lix.systems/lix-project/nix-eval-jobs"
}, },
@ -95,11 +95,11 @@
] ]
}, },
"locked": { "locked": {
"lastModified": 1729742964, "lastModified": 1720066371,
"narHash": "sha256-B4mzTcQ0FZHdpeWcpDYPERtyjJd/NIuaQ9+BV1h+MpA=", "narHash": "sha256-uPlLYH2S0ACj0IcgaK9Lsf4spmJoGejR9DotXiXSBZQ=",
"owner": "nix-community", "owner": "nix-community",
"repo": "nix-github-actions", "repo": "nix-github-actions",
"rev": "e04df33f62cdcf93d73e9a04142464753a16db67", "rev": "622f829f5fe69310a866c8a6cd07e747c44ef820",
"type": "github" "type": "github"
}, },
"original": { "original": {
@ -111,11 +111,11 @@
"nix2container": { "nix2container": {
"flake": false, "flake": false,
"locked": { "locked": {
"lastModified": 1724996935, "lastModified": 1720642556,
"narHash": "sha256-njRK9vvZ1JJsP8oV2OgkBrpJhgQezI03S7gzskCcHos=", "narHash": "sha256-qsnqk13UmREKmRT7c8hEnz26X3GFFyIQrqx4EaRc1Is=",
"owner": "nlewo", "owner": "nlewo",
"repo": "nix2container", "repo": "nix2container",
"rev": "fa6bb0a1159f55d071ba99331355955ae30b3401", "rev": "3853e5caf9ad24103b13aa6e0e8bcebb47649fe4",
"type": "github" "type": "github"
}, },
"original": { "original": {
@ -126,11 +126,11 @@
}, },
"nixpkgs": { "nixpkgs": {
"locked": { "locked": {
"lastModified": 1728193676, "lastModified": 1723688146,
"narHash": "sha256-PbDWAIjKJdlVg+qQRhzdSor04bAPApDqIv2DofTyynk=", "narHash": "sha256-sqLwJcHYeWLOeP/XoLwAtYjr01TISlkOfz+NG82pbdg=",
"owner": "NixOS", "owner": "NixOS",
"repo": "nixpkgs", "repo": "nixpkgs",
"rev": "ecbc1ca8ffd6aea8372ad16be9ebbb39889e55b6", "rev": "c3d4ac725177c030b1e289015989da2ad9d56af0",
"type": "github" "type": "github"
}, },
"original": { "original": {
@ -159,11 +159,11 @@
"pre-commit-hooks": { "pre-commit-hooks": {
"flake": false, "flake": false,
"locked": { "locked": {
"lastModified": 1726745158, "lastModified": 1721042469,
"narHash": "sha256-D5AegvGoEjt4rkKedmxlSEmC+nNLMBPWFxvmYnVLhjk=", "narHash": "sha256-6FPUl7HVtvRHCCBQne7Ylp4p+dpP3P/OYuzjztZ4s70=",
"owner": "cachix", "owner": "cachix",
"repo": "git-hooks.nix", "repo": "git-hooks.nix",
"rev": "4e743a6920eab45e8ba0fbe49dc459f1423a4b74", "rev": "f451c19376071a90d8c58ab1a953c6e9840527fd",
"type": "github" "type": "github"
}, },
"original": { "original": {
@ -187,11 +187,11 @@
] ]
}, },
"locked": { "locked": {
"lastModified": 1730321837, "lastModified": 1723454642,
"narHash": "sha256-vK+a09qq19QNu2MlLcvN4qcRctJbqWkX7ahgPZ/+maI=", "narHash": "sha256-S0Gvsenh0II7EAaoc9158ZB4vYyuycvMGKGxIbERNAM=",
"owner": "numtide", "owner": "numtide",
"repo": "treefmt-nix", "repo": "treefmt-nix",
"rev": "746901bb8dba96d154b66492a29f5db0693dbfcc", "rev": "349de7bc435bdff37785c2466f054ed1766173be",
"type": "github" "type": "github"
}, },
"original": { "original": {

15
flake.nix
View file

@ -73,21 +73,6 @@
default = pkgsBySystem.${system}.hydra; default = pkgsBySystem.${system}.hydra;
}); });
devShells = forEachSystem (system: let
pkgs = pkgsBySystem.${system};
lib = pkgs.lib;
mkDevShell = stdenv: (pkgs.mkShell.override { inherit stdenv; }) {
inputsFrom = [ (self.packages.${system}.default.override { inherit stdenv; }) ];
packages =
lib.optional (stdenv.cc.isClang && stdenv.hostPlatform == stdenv.buildPlatform) pkgs.clang-tools;
};
in {
default = mkDevShell pkgs.stdenv;
clang = mkDevShell pkgs.clangStdenv;
});
nixosModules = import ./nixos-modules { nixosModules = import ./nixos-modules {
overlays = overlayList; overlays = overlayList;
}; };

32
src/hydra-evaluator/hydra-evaluator.cc
View file

@ -14,12 +14,11 @@
#include <sys/wait.h> #include <sys/wait.h>
#include <boost/format.hpp> #include <boost/format.hpp>
#include <utility>
using namespace nix; using namespace nix;
using boost::format; using boost::format;
using JobsetName = std::pair<std::string, std::string>; typedef std::pair<std::string, std::string> JobsetName;
class JobsetId { class JobsetId {
public: public:
@ -29,8 +28,8 @@ class JobsetId {
int id; int id;
JobsetId(std::string project, std::string jobset, int id) JobsetId(const std::string & project, const std::string & jobset, int id)
: project{std::move( project )}, jobset{std::move( jobset )}, id{ id } : project{ project }, jobset{ jobset }, id{ id }
{ {
} }
@ -42,7 +41,7 @@ class JobsetId {
friend bool operator== (const JobsetId & lhs, const JobsetName & rhs); friend bool operator== (const JobsetId & lhs, const JobsetName & rhs);
friend bool operator!= (const JobsetId & lhs, const JobsetName & rhs); friend bool operator!= (const JobsetId & lhs, const JobsetName & rhs);
[[nodiscard]] std::string display() const { std::string display() const {
return str(format("%1%:%2% (jobset#%3%)") % project % jobset % id); return str(format("%1%:%2% (jobset#%3%)") % project % jobset % id);
} }
}; };
@ -89,11 +88,11 @@ struct Evaluator
JobsetId name; JobsetId name;
std::optional<EvaluationStyle> evaluation_style; std::optional<EvaluationStyle> evaluation_style;
time_t lastCheckedTime, triggerTime; time_t lastCheckedTime, triggerTime;
time_t checkInterval; int checkInterval;
Pid pid; Pid pid;
}; };
using Jobsets = std::map<JobsetId, Jobset>; typedef std::map<JobsetId, Jobset> Jobsets;
std::optional<JobsetName> evalOne; std::optional<JobsetName> evalOne;
@ -139,15 +138,13 @@ struct Evaluator
if (evalOne && name != *evalOne) continue; if (evalOne && name != *evalOne) continue;
auto res = state->jobsets.try_emplace(name, Jobset{.name=name}); auto res = state->jobsets.try_emplace(name, Jobset{name});
auto & jobset = res.first->second; auto & jobset = res.first->second;
jobset.lastCheckedTime = row["lastCheckedTime"].as<time_t>(0); jobset.lastCheckedTime = row["lastCheckedTime"].as<time_t>(0);
jobset.triggerTime = row["triggerTime"].as<time_t>(notTriggered); jobset.triggerTime = row["triggerTime"].as<time_t>(notTriggered);
jobset.checkInterval = row["checkInterval"].as<time_t>(); jobset.checkInterval = row["checkInterval"].as<time_t>();
switch (row["jobset_enabled"].as<int>(0)) {
int eval_style = row["jobset_enabled"].as<int>(0);
switch (eval_style) {
case 1: case 1:
jobset.evaluation_style = EvaluationStyle::SCHEDULE; jobset.evaluation_style = EvaluationStyle::SCHEDULE;
break; break;
@ -157,9 +154,6 @@ struct Evaluator
case 3: case 3:
jobset.evaluation_style = EvaluationStyle::ONE_AT_A_TIME; jobset.evaluation_style = EvaluationStyle::ONE_AT_A_TIME;
break; break;
default:
// Disabled or unknown. Leave as nullopt.
break;
} }
seen.insert(name); seen.insert(name);
@ -181,7 +175,7 @@ struct Evaluator
void startEval(State & state, Jobset & jobset) void startEval(State & state, Jobset & jobset)
{ {
time_t now = time(nullptr); time_t now = time(0);
printInfo("starting evaluation of jobset ‘%s’ (last checked %d s ago)", printInfo("starting evaluation of jobset ‘%s’ (last checked %d s ago)",
jobset.name.display(), jobset.name.display(),
@ -234,7 +228,7 @@ struct Evaluator
return false; return false;
} }
if (jobset.lastCheckedTime + jobset.checkInterval <= time(nullptr)) { if (jobset.lastCheckedTime + jobset.checkInterval <= time(0)) {
// Time to schedule a fresh evaluation. If the jobset // Time to schedule a fresh evaluation. If the jobset
// is a ONE_AT_A_TIME jobset, ensure the previous jobset // is a ONE_AT_A_TIME jobset, ensure the previous jobset
// has no remaining, unfinished work. // has no remaining, unfinished work.
@ -307,7 +301,7 @@ struct Evaluator
/* Put jobsets in order of ascending trigger time, last checked /* Put jobsets in order of ascending trigger time, last checked
time, and name. */ time, and name. */
std::ranges::sort(sorted, std::sort(sorted.begin(), sorted.end(),
[](const Jobsets::iterator & a, const Jobsets::iterator & b) { [](const Jobsets::iterator & a, const Jobsets::iterator & b) {
return return
a->second.triggerTime != b->second.triggerTime a->second.triggerTime != b->second.triggerTime
@ -330,7 +324,7 @@ struct Evaluator
while (true) { while (true) {
time_t now = time(nullptr); time_t now = time(0);
std::chrono::seconds sleepTime = std::chrono::seconds::max(); std::chrono::seconds sleepTime = std::chrono::seconds::max();
@ -417,7 +411,7 @@ struct Evaluator
printInfo("evaluation of jobset ‘%s’ %s", printInfo("evaluation of jobset ‘%s’ %s",
jobset.name.display(), statusToString(status)); jobset.name.display(), statusToString(status));
auto now = time(nullptr); auto now = time(0);
jobset.triggerTime = notTriggered; jobset.triggerTime = notTriggered;
jobset.lastCheckedTime = now; jobset.lastCheckedTime = now;

20
src/hydra-queue-runner/build-remote.cc
View file

@ -1,6 +1,5 @@
#include <algorithm> #include <algorithm>
#include <cmath> #include <cmath>
#include <ranges>
#include <sys/types.h> #include <sys/types.h>
#include <sys/stat.h> #include <sys/stat.h>
@ -14,7 +13,6 @@
#include "serve-protocol.hh" #include "serve-protocol.hh"
#include "serve-protocol-impl.hh" #include "serve-protocol-impl.hh"
#include "ssh.hh" #include "ssh.hh"
#include "temporary-dir.hh"
#include "finally.hh" #include "finally.hh"
#include "url.hh" #include "url.hh"
@ -43,7 +41,6 @@ static Strings extraStoreArgs(std::string & machine)
} }
} catch (BadURL &) { } catch (BadURL &) {
// We just try to continue with `machine->sshName` here for backwards compat. // We just try to continue with `machine->sshName` here for backwards compat.
printMsg(lvlWarn, "could not parse machine URL '%s', passing through to SSH", machine);
} }
return result; return result;
@ -136,8 +133,8 @@ static void copyClosureTo(
auto sorted = destStore.topoSortPaths(closure); auto sorted = destStore.topoSortPaths(closure);
StorePathSet missing; StorePathSet missing;
for (auto & i : std::ranges::reverse_view(sorted)) for (auto i = sorted.rbegin(); i != sorted.rend(); ++i)
if (!present.count(i)) missing.insert(i); if (!present.count(*i)) missing.insert(*i);
printMsg(lvlDebug, "sending %d missing paths", missing.size()); printMsg(lvlDebug, "sending %d missing paths", missing.size());
@ -307,12 +304,12 @@ static BuildResult performBuild(
time_t startTime, stopTime; time_t startTime, stopTime;
startTime = time(nullptr); startTime = time(0);
{ {
MaintainCount<counter> mc(nrStepsBuilding); MaintainCount<counter> mc(nrStepsBuilding);
result = ServeProto::Serialise<BuildResult>::read(localStore, conn); result = ServeProto::Serialise<BuildResult>::read(localStore, conn);
} }
stopTime = time(nullptr); stopTime = time(0);
if (!result.startTime) { if (!result.startTime) {
// If the builder gave `startTime = 0`, use our measurements // If the builder gave `startTime = 0`, use our measurements
@ -341,10 +338,10 @@ static BuildResult performBuild(
// were known // were known
assert(outputPath); assert(outputPath);
auto outputHash = outputHashes.at(outputName); auto outputHash = outputHashes.at(outputName);
auto drvOutput = DrvOutput { .drvHash=outputHash, .outputName=outputName }; auto drvOutput = DrvOutput { outputHash, outputName };
result.builtOutputs.insert_or_assign( result.builtOutputs.insert_or_assign(
std::move(outputName), std::move(outputName),
Realisation { .id=drvOutput, .outPath=*outputPath }); Realisation { drvOutput, *outputPath });
} }
} }
@ -626,7 +623,6 @@ void State::buildRemote(ref<Store> destStore,
/* Throttle CPU-bound work. Opportunistically skip updating the current /* Throttle CPU-bound work. Opportunistically skip updating the current
* step, since this requires a DB roundtrip. */ * step, since this requires a DB roundtrip. */
if (!localWorkThrottler.try_acquire()) { if (!localWorkThrottler.try_acquire()) {
MaintainCount<counter> mc(nrStepsWaitingForDownloadSlot);
updateStep(ssWaitingForLocalSlot); updateStep(ssWaitingForLocalSlot);
localWorkThrottler.acquire(); localWorkThrottler.acquire();
} }
@ -638,7 +634,7 @@ void State::buildRemote(ref<Store> destStore,
* copying outputs and we end up building too many things that we * copying outputs and we end up building too many things that we
* haven't been able to allow copy slots for. */ * haven't been able to allow copy slots for. */
assert(reservation.unique()); assert(reservation.unique());
reservation = nullptr; reservation = 0;
wakeDispatcher(); wakeDispatcher();
StorePathSet outputs; StorePathSet outputs;
@ -701,7 +697,7 @@ void State::buildRemote(ref<Store> destStore,
if (info->consecutiveFailures == 0 || info->lastFailure < now - std::chrono::seconds(30)) { if (info->consecutiveFailures == 0 || info->lastFailure < now - std::chrono::seconds(30)) {
info->consecutiveFailures = std::min(info->consecutiveFailures + 1, (unsigned int) 4); info->consecutiveFailures = std::min(info->consecutiveFailures + 1, (unsigned int) 4);
info->lastFailure = now; info->lastFailure = now;
int delta = static_cast<int>(retryInterval * std::pow(retryBackoff, info->consecutiveFailures - 1) + (rand() % 30)); int delta = retryInterval * std::pow(retryBackoff, info->consecutiveFailures - 1) + (rand() % 30);
printMsg(lvlInfo, "will disable machine ‘%1%’ for %2%s", machine->sshName, delta); printMsg(lvlInfo, "will disable machine ‘%1%’ for %2%s", machine->sshName, delta);
info->disabledUntil = now + std::chrono::seconds(delta); info->disabledUntil = now + std::chrono::seconds(delta);
} }

1
src/hydra-queue-runner/build-result.cc
View file

@ -1,7 +1,6 @@
#include "hydra-build-result.hh" #include "hydra-build-result.hh"
#include "store-api.hh" #include "store-api.hh"
#include "fs-accessor.hh" #include "fs-accessor.hh"
#include "strings.hh"
#include <regex> #include <regex>

13
src/hydra-queue-runner/builder.cc
View file

@ -1,6 +1,5 @@
#include <cmath> #include <cmath>
#include "error.hh"
#include "state.hh" #include "state.hh"
#include "hydra-build-result.hh" #include "hydra-build-result.hh"
#include "finally.hh" #include "finally.hh"
@ -59,7 +58,7 @@ void State::builder(MachineReservation::ptr reservation)
/* If the machine hasn't been released yet, release and wake up the dispatcher. */ /* If the machine hasn't been released yet, release and wake up the dispatcher. */
if (reservation) { if (reservation) {
assert(reservation.unique()); assert(reservation.unique());
reservation = nullptr; reservation = 0;
wakeDispatcher(); wakeDispatcher();
} }
@ -73,7 +72,7 @@ void State::builder(MachineReservation::ptr reservation)
step_->tries++; step_->tries++;
nrRetries++; nrRetries++;
if (step_->tries > maxNrRetries) maxNrRetries = step_->tries; // yeah yeah, not atomic if (step_->tries > maxNrRetries) maxNrRetries = step_->tries; // yeah yeah, not atomic
int delta = static_cast<int>(retryInterval * std::pow(retryBackoff, step_->tries - 1) + (rand() % 10)); int delta = retryInterval * std::pow(retryBackoff, step_->tries - 1) + (rand() % 10);
printMsg(lvlInfo, "will retry ‘%s’ after %ss", localStore->printStorePath(step->drvPath), delta); printMsg(lvlInfo, "will retry ‘%s’ after %ss", localStore->printStorePath(step->drvPath), delta);
step_->after = std::chrono::system_clock::now() + std::chrono::seconds(delta); step_->after = std::chrono::system_clock::now() + std::chrono::seconds(delta);
} }
@ -186,12 +185,12 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
unlink(result.logFile.c_str()); unlink(result.logFile.c_str());
} }
} catch (...) { } catch (...) {
ignoreExceptionInDestructor(); ignoreException();
} }
} }
}); });
time_t stepStartTime = result.startTime = time(nullptr); time_t stepStartTime = result.startTime = time(0);
/* If any of the outputs have previously failed, then don't bother /* If any of the outputs have previously failed, then don't bother
building again. */ building again. */
@ -238,7 +237,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
} }
} }
time_t stepStopTime = time(nullptr); time_t stepStopTime = time(0);
if (!result.stopTime) result.stopTime = stepStopTime; if (!result.stopTime) result.stopTime = stepStopTime;
/* For standard failures, we don't care about the error /* For standard failures, we don't care about the error
@ -252,7 +251,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
auto step_(step->state.lock()); auto step_(step->state.lock());
if (!step_->jobsets.empty()) { if (!step_->jobsets.empty()) {
// FIXME: loss of precision. // FIXME: loss of precision.
time_t charge = (result.stopTime - result.startTime) / static_cast<time_t>(step_->jobsets.size()); time_t charge = (result.stopTime - result.startTime) / step_->jobsets.size();
for (auto & jobset : step_->jobsets) for (auto & jobset : step_->jobsets)
jobset->addStep(result.startTime, charge); jobset->addStep(result.startTime, charge);
} }

18
src/hydra-queue-runner/dispatcher.cc
View file

@ -46,7 +46,7 @@ void State::dispatcher()
auto t_after_work = std::chrono::steady_clock::now(); auto t_after_work = std::chrono::steady_clock::now();
prom.dispatcher_time_spent_running.Increment( prom.dispatcher_time_spent_running.Increment(
static_cast<double>(std::chrono::duration_cast<std::chrono::microseconds>(t_after_work - t_before_work).count())); std::chrono::duration_cast<std::chrono::microseconds>(t_after_work - t_before_work).count());
dispatchTimeMs += std::chrono::duration_cast<std::chrono::milliseconds>(t_after_work - t_before_work).count(); dispatchTimeMs += std::chrono::duration_cast<std::chrono::milliseconds>(t_after_work - t_before_work).count();
/* Sleep until we're woken up (either because a runnable build /* Sleep until we're woken up (either because a runnable build
@ -63,7 +63,7 @@ void State::dispatcher()
auto t_after_sleep = std::chrono::steady_clock::now(); auto t_after_sleep = std::chrono::steady_clock::now();
prom.dispatcher_time_spent_waiting.Increment( prom.dispatcher_time_spent_waiting.Increment(
static_cast<double>(std::chrono::duration_cast<std::chrono::microseconds>(t_after_sleep - t_after_work).count())); std::chrono::duration_cast<std::chrono::microseconds>(t_after_sleep - t_after_work).count());
} catch (std::exception & e) { } catch (std::exception & e) {
printError("dispatcher: %s", e.what()); printError("dispatcher: %s", e.what());
@ -190,7 +190,7 @@ system_time State::doDispatch()
} }
} }
std::ranges::sort(runnableSorted, sort(runnableSorted.begin(), runnableSorted.end(),
[](const StepInfo & a, const StepInfo & b) [](const StepInfo & a, const StepInfo & b)
{ {
return return
@ -240,11 +240,11 @@ system_time State::doDispatch()
- Then by speed factor. - Then by speed factor.
- Finally by load. */ - Finally by load. */
std::ranges::sort(machinesSorted, sort(machinesSorted.begin(), machinesSorted.end(),
[](const MachineInfo & a, const MachineInfo & b) -> bool [](const MachineInfo & a, const MachineInfo & b) -> bool
{ {
float ta = std::round(static_cast<float>(a.currentJobs) / a.machine->speedFactorFloat); float ta = std::round(a.currentJobs / a.machine->speedFactorFloat);
float tb = std::round(static_cast<float>(b.currentJobs) / b.machine->speedFactorFloat); float tb = std::round(b.currentJobs / b.machine->speedFactorFloat);
return return
ta != tb ? ta < tb : ta != tb ? ta < tb :
a.machine->speedFactorFloat != b.machine->speedFactorFloat ? a.machine->speedFactorFloat > b.machine->speedFactorFloat : a.machine->speedFactorFloat != b.machine->speedFactorFloat ? a.machine->speedFactorFloat > b.machine->speedFactorFloat :
@ -345,7 +345,7 @@ void State::abortUnsupported()
auto machines2 = *machines.lock(); auto machines2 = *machines.lock();
system_time now = std::chrono::system_clock::now(); system_time now = std::chrono::system_clock::now();
auto now2 = time(nullptr); auto now2 = time(0);
std::unordered_set<Step::ptr> aborted; std::unordered_set<Step::ptr> aborted;
@ -436,7 +436,7 @@ void Jobset::addStep(time_t startTime, time_t duration)
void Jobset::pruneSteps() void Jobset::pruneSteps()
{ {
time_t now = time(nullptr); time_t now = time(0);
auto steps_(steps.lock()); auto steps_(steps.lock());
while (!steps_->empty()) { while (!steps_->empty()) {
auto i = steps_->begin(); auto i = steps_->begin();
@ -464,7 +464,7 @@ State::MachineReservation::~MachineReservation()
auto prev = machine->state->currentJobs--; auto prev = machine->state->currentJobs--;
assert(prev); assert(prev);
if (prev == 1) if (prev == 1)
machine->state->idleSince = time(nullptr); machine->state->idleSince = time(0);
{ {
auto machineTypes_(state.machineTypes.lock()); auto machineTypes_(state.machineTypes.lock());

2
src/hydra-queue-runner/hydra-build-result.hh
View file

@ -14,7 +14,7 @@ struct BuildProduct
bool isRegular = false; bool isRegular = false;
std::optional<nix::Hash> sha256hash; std::optional<nix::Hash> sha256hash;
std::optional<off_t> fileSize; std::optional<off_t> fileSize;
BuildProduct() = default; BuildProduct() { }
}; };
struct BuildMetric struct BuildMetric

51
src/hydra-queue-runner/hydra-queue-runner.cc
View file

@ -105,7 +105,7 @@ State::State(std::optional<std::string> metricsAddrOpt)
: config(std::make_unique<HydraConfig>()) : config(std::make_unique<HydraConfig>())
, maxUnsupportedTime(config->getIntOption("max_unsupported_time", 0)) , maxUnsupportedTime(config->getIntOption("max_unsupported_time", 0))
, dbPool(config->getIntOption("max_db_connections", 128)) , dbPool(config->getIntOption("max_db_connections", 128))
, localWorkThrottler(static_cast<ptrdiff_t>(config->getIntOption("max_local_worker_threads", std::min(maxSupportedLocalWorkers, std::max(4u, std::thread::hardware_concurrency()) - 2)))) , localWorkThrottler(config->getIntOption("max_local_worker_threads", std::min(maxSupportedLocalWorkers, std::max(4u, std::thread::hardware_concurrency()) - 2)))
, maxOutputSize(config->getIntOption("max_output_size", 2ULL << 30)) , maxOutputSize(config->getIntOption("max_output_size", 2ULL << 30))
, maxLogSize(config->getIntOption("max_log_size", 64ULL << 20)) , maxLogSize(config->getIntOption("max_log_size", 64ULL << 20))
, uploadLogsToBinaryCache(config->getBoolOption("upload_logs_to_binary_cache", false)) , uploadLogsToBinaryCache(config->getBoolOption("upload_logs_to_binary_cache", false))
@ -138,7 +138,7 @@ nix::MaintainCount<counter> State::startDbUpdate()
{ {
if (nrActiveDbUpdates > 6) if (nrActiveDbUpdates > 6)
printError("warning: %d concurrent database updates; PostgreSQL may be stalled", nrActiveDbUpdates.load()); printError("warning: %d concurrent database updates; PostgreSQL may be stalled", nrActiveDbUpdates.load());
return {nrActiveDbUpdates}; return MaintainCount<counter>(nrActiveDbUpdates);
} }
@ -171,9 +171,9 @@ void State::parseMachines(const std::string & contents)
for (auto & f : mandatoryFeatures) for (auto & f : mandatoryFeatures)
supportedFeatures.insert(f); supportedFeatures.insert(f);
using MaxJobs = std::remove_const_t<decltype(nix::Machine::maxJobs)>; using MaxJobs = std::remove_const<decltype(nix::Machine::maxJobs)>::type;
auto machine = std::make_shared<::Machine>(::Machine {{ auto machine = std::make_shared<::Machine>(nix::Machine {
// `storeUri`, not yet used // `storeUri`, not yet used
"", "",
// `systemTypes`, not yet used // `systemTypes`, not yet used
@ -194,11 +194,11 @@ void State::parseMachines(const std::string & contents)
tokens[7] != "" && tokens[7] != "-" tokens[7] != "" && tokens[7] != "-"
? base64Decode(tokens[7]) ? base64Decode(tokens[7])
: "", : "",
}}); });
machine->sshName = tokens[0]; machine->sshName = tokens[0];
machine->systemTypesSet = tokenizeString<StringSet>(tokens[1], ","); machine->systemTypesSet = tokenizeString<StringSet>(tokens[1], ",");
machine->speedFactorFloat = static_cast<float>(atof(tokens[4].c_str())); machine->speedFactorFloat = atof(tokens[4].c_str());
/* Re-use the State object of the previous machine with the /* Re-use the State object of the previous machine with the
same name. */ same name. */
@ -412,7 +412,7 @@ void State::finishBuildStep(pqxx::work & txn, const RemoteResult & result,
} }
unsigned int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime, int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime,
Build::ptr build, const StorePath & drvPath, const nix::Derivation drv, const std::string & outputName, const StorePath & storePath) Build::ptr build, const StorePath & drvPath, const nix::Derivation drv, const std::string & outputName, const StorePath & storePath)
{ {
restart: restart:
@ -594,7 +594,7 @@ std::shared_ptr<PathLocks> State::acquireGlobalLock()
createDirs(dirOf(lockPath)); createDirs(dirOf(lockPath));
auto lock = std::make_shared<PathLocks>(); auto lock = std::make_shared<PathLocks>();
if (!lock->lockPaths(PathSet({lockPath}), "", false)) return nullptr; if (!lock->lockPaths(PathSet({lockPath}), "", false)) return 0;
return lock; return lock;
} }
@ -602,10 +602,10 @@ std::shared_ptr<PathLocks> State::acquireGlobalLock()
void State::dumpStatus(Connection & conn) void State::dumpStatus(Connection & conn)
{ {
time_t now = time(nullptr); time_t now = time(0);
json statusJson = { json statusJson = {
{"status", "up"}, {"status", "up"},
{"time", time(nullptr)}, {"time", time(0)},
{"uptime", now - startedAt}, {"uptime", now - startedAt},
{"pid", getpid()}, {"pid", getpid()},
@ -613,7 +613,6 @@ void State::dumpStatus(Connection & conn)
{"nrActiveSteps", activeSteps_.lock()->size()}, {"nrActiveSteps", activeSteps_.lock()->size()},
{"nrStepsBuilding", nrStepsBuilding.load()}, {"nrStepsBuilding", nrStepsBuilding.load()},
{"nrStepsCopyingTo", nrStepsCopyingTo.load()}, {"nrStepsCopyingTo", nrStepsCopyingTo.load()},
{"nrStepsWaitingForDownloadSlot", nrStepsWaitingForDownloadSlot.load()},
{"nrStepsCopyingFrom", nrStepsCopyingFrom.load()}, {"nrStepsCopyingFrom", nrStepsCopyingFrom.load()},
{"nrStepsWaiting", nrStepsWaiting.load()}, {"nrStepsWaiting", nrStepsWaiting.load()},
{"nrUnsupportedSteps", nrUnsupportedSteps.load()}, {"nrUnsupportedSteps", nrUnsupportedSteps.load()},
@ -621,7 +620,7 @@ void State::dumpStatus(Connection & conn)
{"bytesReceived", bytesReceived.load()}, {"bytesReceived", bytesReceived.load()},
{"nrBuildsRead", nrBuildsRead.load()}, {"nrBuildsRead", nrBuildsRead.load()},
{"buildReadTimeMs", buildReadTimeMs.load()}, {"buildReadTimeMs", buildReadTimeMs.load()},
{"buildReadTimeAvgMs", nrBuildsRead == 0 ? 0.0 : (float) buildReadTimeMs / (float) nrBuildsRead}, {"buildReadTimeAvgMs", nrBuildsRead == 0 ? 0.0 : (float) buildReadTimeMs / nrBuildsRead},
{"nrBuildsDone", nrBuildsDone.load()}, {"nrBuildsDone", nrBuildsDone.load()},
{"nrStepsStarted", nrStepsStarted.load()}, {"nrStepsStarted", nrStepsStarted.load()},
{"nrStepsDone", nrStepsDone.load()}, {"nrStepsDone", nrStepsDone.load()},
@ -630,7 +629,7 @@ void State::dumpStatus(Connection & conn)
{"nrQueueWakeups", nrQueueWakeups.load()}, {"nrQueueWakeups", nrQueueWakeups.load()},
{"nrDispatcherWakeups", nrDispatcherWakeups.load()}, {"nrDispatcherWakeups", nrDispatcherWakeups.load()},
{"dispatchTimeMs", dispatchTimeMs.load()}, {"dispatchTimeMs", dispatchTimeMs.load()},
{"dispatchTimeAvgMs", nrDispatcherWakeups == 0 ? 0.0 : (float) dispatchTimeMs / (float) nrDispatcherWakeups}, {"dispatchTimeAvgMs", nrDispatcherWakeups == 0 ? 0.0 : (float) dispatchTimeMs / nrDispatcherWakeups},
{"nrDbConnections", dbPool.count()}, {"nrDbConnections", dbPool.count()},
{"nrActiveDbUpdates", nrActiveDbUpdates.load()}, {"nrActiveDbUpdates", nrActiveDbUpdates.load()},
}; };
@ -650,8 +649,8 @@ void State::dumpStatus(Connection & conn)
if (nrStepsDone) { if (nrStepsDone) {
statusJson["totalStepTime"] = totalStepTime.load(); statusJson["totalStepTime"] = totalStepTime.load();
statusJson["totalStepBuildTime"] = totalStepBuildTime.load(); statusJson["totalStepBuildTime"] = totalStepBuildTime.load();
statusJson["avgStepTime"] = (float) totalStepTime / (float) nrStepsDone; statusJson["avgStepTime"] = (float) totalStepTime / nrStepsDone;
statusJson["avgStepBuildTime"] = (float) totalStepBuildTime / (float) nrStepsDone; statusJson["avgStepBuildTime"] = (float) totalStepBuildTime / nrStepsDone;
} }
{ {
@ -678,8 +677,8 @@ void State::dumpStatus(Connection & conn)
if (m->state->nrStepsDone) { if (m->state->nrStepsDone) {
machine["totalStepTime"] = s->totalStepTime.load(); machine["totalStepTime"] = s->totalStepTime.load();
machine["totalStepBuildTime"] = s->totalStepBuildTime.load(); machine["totalStepBuildTime"] = s->totalStepBuildTime.load();
machine["avgStepTime"] = (float) s->totalStepTime / (float) s->nrStepsDone; machine["avgStepTime"] = (float) s->totalStepTime / s->nrStepsDone;
machine["avgStepBuildTime"] = (float) s->totalStepBuildTime / (float) s->nrStepsDone; machine["avgStepBuildTime"] = (float) s->totalStepBuildTime / s->nrStepsDone;
} }
statusJson["machines"][m->sshName] = machine; statusJson["machines"][m->sshName] = machine;
} }
@ -707,7 +706,7 @@ void State::dumpStatus(Connection & conn)
}; };
if (i.second.runnable > 0) if (i.second.runnable > 0)
machineTypeJson["waitTime"] = i.second.waitTime.count() + machineTypeJson["waitTime"] = i.second.waitTime.count() +
i.second.runnable * (time(nullptr) - lastDispatcherCheck); i.second.runnable * (time(0) - lastDispatcherCheck);
if (i.second.running == 0) if (i.second.running == 0)
machineTypeJson["lastActive"] = std::chrono::system_clock::to_time_t(i.second.lastActive); machineTypeJson["lastActive"] = std::chrono::system_clock::to_time_t(i.second.lastActive);
} }
@ -733,11 +732,11 @@ void State::dumpStatus(Connection & conn)
{"narWriteCompressionTimeMs", stats.narWriteCompressionTimeMs.load()}, {"narWriteCompressionTimeMs", stats.narWriteCompressionTimeMs.load()},
{"narCompressionSavings", {"narCompressionSavings",
stats.narWriteBytes stats.narWriteBytes
? 1.0 - (double) stats.narWriteCompressedBytes / (double) stats.narWriteBytes ? 1.0 - (double) stats.narWriteCompressedBytes / stats.narWriteBytes
: 0.0}, : 0.0},
{"narCompressionSpeed", // MiB/s {"narCompressionSpeed", // MiB/s
stats.narWriteCompressionTimeMs stats.narWriteCompressionTimeMs
? (double) stats.narWriteBytes / (double) stats.narWriteCompressionTimeMs * 1000.0 / (1024.0 * 1024.0) ? (double) stats.narWriteBytes / stats.narWriteCompressionTimeMs * 1000.0 / (1024.0 * 1024.0)
: 0.0}, : 0.0},
}; };
@ -750,20 +749,20 @@ void State::dumpStatus(Connection & conn)
{"putTimeMs", s3Stats.putTimeMs.load()}, {"putTimeMs", s3Stats.putTimeMs.load()},
{"putSpeed", {"putSpeed",
s3Stats.putTimeMs s3Stats.putTimeMs
? (double) s3Stats.putBytes / (double) s3Stats.putTimeMs * 1000.0 / (1024.0 * 1024.0) ? (double) s3Stats.putBytes / s3Stats.putTimeMs * 1000.0 / (1024.0 * 1024.0)
: 0.0}, : 0.0},
{"get", s3Stats.get.load()}, {"get", s3Stats.get.load()},
{"getBytes", s3Stats.getBytes.load()}, {"getBytes", s3Stats.getBytes.load()},
{"getTimeMs", s3Stats.getTimeMs.load()}, {"getTimeMs", s3Stats.getTimeMs.load()},
{"getSpeed", {"getSpeed",
s3Stats.getTimeMs s3Stats.getTimeMs
? (double) s3Stats.getBytes / (double) s3Stats.getTimeMs * 1000.0 / (1024.0 * 1024.0) ? (double) s3Stats.getBytes / s3Stats.getTimeMs * 1000.0 / (1024.0 * 1024.0)
: 0.0}, : 0.0},
{"head", s3Stats.head.load()}, {"head", s3Stats.head.load()},
{"costDollarApprox", {"costDollarApprox",
(double) (s3Stats.get + s3Stats.head) / 10000.0 * 0.004 (s3Stats.get + s3Stats.head) / 10000.0 * 0.004
+ (double) s3Stats.put / 1000.0 * 0.005 + + s3Stats.put / 1000.0 * 0.005 +
+ (double) s3Stats.getBytes / (1024.0 * 1024.0 * 1024.0) * 0.09}, + s3Stats.getBytes / (1024.0 * 1024.0 * 1024.0) * 0.09},
}; };
} }
} }
@ -849,7 +848,7 @@ void State::run(BuildID buildOne)
/* Can't be bothered to shut down cleanly. Goodbye! */ /* Can't be bothered to shut down cleanly. Goodbye! */
auto callback = createInterruptCallback([&]() { std::_Exit(0); }); auto callback = createInterruptCallback([&]() { std::_Exit(0); });
startedAt = time(nullptr); startedAt = time(0);
this->buildOne = buildOne; this->buildOne = buildOne;
auto lock = acquireGlobalLock(); auto lock = acquireGlobalLock();

69
src/hydra-queue-runner/nar-extractor.cc
View file

@ -3,41 +3,11 @@
#include "archive.hh" #include "archive.hh"
#include <unordered_set> #include <unordered_set>
#include <utility>
using namespace nix; using namespace nix;
struct Extractor : NARParseVisitor struct Extractor : ParseSink
{ {
class MyFileHandle : public FileHandle
{
NarMemberData & memberData;
uint64_t expectedSize;
std::unique_ptr<HashSink> hashSink;
public:
MyFileHandle(NarMemberData & memberData, uint64_t size) : memberData(memberData), expectedSize(size)
{
hashSink = std::make_unique<HashSink>(HashType::SHA256);
}
void receiveContents(std::string_view data) override
{
*memberData.fileSize += data.size();
(*hashSink)(data);
if (memberData.contents) {
memberData.contents->append(data);
}
assert(memberData.fileSize <= expectedSize);
if (memberData.fileSize == expectedSize) {
auto [hash, len] = hashSink->finish();
assert(memberData.fileSize == len);
memberData.sha256 = hash;
hashSink.reset();
}
}
};
std::unordered_set<Path> filesToKeep { std::unordered_set<Path> filesToKeep {
"/nix-support/hydra-build-products", "/nix-support/hydra-build-products",
"/nix-support/hydra-release-name", "/nix-support/hydra-release-name",
@ -45,10 +15,11 @@ struct Extractor : NARParseVisitor
}; };
NarMemberDatas & members; NarMemberDatas & members;
NarMemberData * curMember = nullptr;
Path prefix; Path prefix;
Extractor(NarMemberDatas & members, Path prefix) Extractor(NarMemberDatas & members, const Path & prefix)
: members(members), prefix(std::move(prefix)) : members(members), prefix(prefix)
{ } { }
void createDirectory(const Path & path) override void createDirectory(const Path & path) override
@ -56,15 +27,41 @@ struct Extractor : NARParseVisitor
members.insert_or_assign(prefix + path, NarMemberData { .type = FSAccessor::Type::tDirectory }); members.insert_or_assign(prefix + path, NarMemberData { .type = FSAccessor::Type::tDirectory });
} }
std::unique_ptr<FileHandle> createRegularFile(const Path & path, uint64_t size, bool executable) override void createRegularFile(const Path & path) override
{ {
auto memberData = &members.insert_or_assign(prefix + path, NarMemberData { curMember = &members.insert_or_assign(prefix + path, NarMemberData {
.type = FSAccessor::Type::tRegular, .type = FSAccessor::Type::tRegular,
.fileSize = 0, .fileSize = 0,
.contents = filesToKeep.count(path) ? std::optional("") : std::nullopt, .contents = filesToKeep.count(path) ? std::optional("") : std::nullopt,
}).first->second; }).first->second;
}
return std::make_unique<MyFileHandle>(*memberData, size); std::optional<uint64_t> expectedSize;
std::unique_ptr<HashSink> hashSink;
void preallocateContents(uint64_t size) override
{
expectedSize = size;
hashSink = std::make_unique<HashSink>(HashType::SHA256);
}
void receiveContents(std::string_view data) override
{
assert(expectedSize);
assert(curMember);
assert(hashSink);
*curMember->fileSize += data.size();
(*hashSink)(data);
if (curMember->contents) {
curMember->contents->append(data);
}
assert(curMember->fileSize <= expectedSize);
if (curMember->fileSize == expectedSize) {
auto [hash, len] = hashSink->finish();
assert(curMember->fileSize == len);
curMember->sha256 = hash;
hashSink.reset();
}
} }
void createSymlink(const Path & path, const std::string & target) override void createSymlink(const Path & path, const std::string & target) override

2
src/hydra-queue-runner/nar-extractor.hh
View file

@ -13,7 +13,7 @@ struct NarMemberData
std::optional<nix::Hash> sha256; std::optional<nix::Hash> sha256;
}; };
using NarMemberDatas = std::map<nix::Path, NarMemberData>; typedef std::map<nix::Path, NarMemberData> NarMemberDatas;
/* Read a NAR from a source and get to some info about every file /* Read a NAR from a source and get to some info about every file
inside the NAR. */ inside the NAR. */

43
src/hydra-queue-runner/queue-monitor.cc
View file

@ -4,8 +4,7 @@
#include "thread-pool.hh" #include "thread-pool.hh"
#include <cstring> #include <cstring>
#include <utility> #include <signal.h>
#include <csignal>
using namespace nix; using namespace nix;
@ -53,7 +52,7 @@ void State::queueMonitorLoop(Connection & conn)
auto t_after_work = std::chrono::steady_clock::now(); auto t_after_work = std::chrono::steady_clock::now();
prom.queue_monitor_time_spent_running.Increment( prom.queue_monitor_time_spent_running.Increment(
static_cast<double>(std::chrono::duration_cast<std::chrono::microseconds>(t_after_work - t_before_work).count())); std::chrono::duration_cast<std::chrono::microseconds>(t_after_work - t_before_work).count());
/* Sleep until we get notification from the database about an /* Sleep until we get notification from the database about an
event. */ event. */
@ -80,7 +79,7 @@ void State::queueMonitorLoop(Connection & conn)
auto t_after_sleep = std::chrono::steady_clock::now(); auto t_after_sleep = std::chrono::steady_clock::now();
prom.queue_monitor_time_spent_waiting.Increment( prom.queue_monitor_time_spent_waiting.Increment(
static_cast<double>(std::chrono::duration_cast<std::chrono::microseconds>(t_after_sleep - t_after_work).count())); std::chrono::duration_cast<std::chrono::microseconds>(t_after_sleep - t_after_work).count());
} }
exit(0); exit(0);
@ -89,7 +88,7 @@ void State::queueMonitorLoop(Connection & conn)
struct PreviousFailure : public std::exception { struct PreviousFailure : public std::exception {
Step::ptr step; Step::ptr step;
PreviousFailure(Step::ptr step) : step(std::move(step)) { } PreviousFailure(Step::ptr step) : step(step) { }
}; };
@ -118,7 +117,7 @@ bool State::getQueuedBuilds(Connection & conn,
for (auto const & row : res) { for (auto const & row : res) {
auto builds_(builds.lock()); auto builds_(builds.lock());
auto id = row["id"].as<BuildID>(); BuildID id = row["id"].as<BuildID>();
if (buildOne && id != buildOne) continue; if (buildOne && id != buildOne) continue;
if (builds_->count(id)) continue; if (builds_->count(id)) continue;
@ -138,7 +137,7 @@ bool State::getQueuedBuilds(Connection & conn,
newIDs.push_back(id); newIDs.push_back(id);
newBuildsByID[id] = build; newBuildsByID[id] = build;
newBuildsByPath.emplace(build->drvPath, id); newBuildsByPath.emplace(std::make_pair(build->drvPath, id));
} }
} }
@ -163,7 +162,7 @@ bool State::getQueuedBuilds(Connection & conn,
("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $3 where id = $1 and finished = 0", ("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $3 where id = $1 and finished = 0",
build->id, build->id,
(int) bsAborted, (int) bsAborted,
time(nullptr)); time(0));
txn.commit(); txn.commit();
build->finishedInDB = true; build->finishedInDB = true;
nrBuildsDone++; nrBuildsDone++;
@ -177,7 +176,7 @@ bool State::getQueuedBuilds(Connection & conn,
/* Create steps for this derivation and its dependencies. */ /* Create steps for this derivation and its dependencies. */
try { try {
step = createStep(destStore, conn, build, build->drvPath, step = createStep(destStore, conn, build, build->drvPath,
build, nullptr, finishedDrvs, newSteps, newRunnable); build, 0, finishedDrvs, newSteps, newRunnable);
} catch (PreviousFailure & ex) { } catch (PreviousFailure & ex) {
/* Some step previously failed, so mark the build as /* Some step previously failed, so mark the build as
@ -222,7 +221,7 @@ bool State::getQueuedBuilds(Connection & conn,
"where id = $1 and finished = 0", "where id = $1 and finished = 0",
build->id, build->id,
(int) (ex.step->drvPath == build->drvPath ? bsFailed : bsDepFailed), (int) (ex.step->drvPath == build->drvPath ? bsFailed : bsDepFailed),
time(nullptr)); time(0));
notifyBuildFinished(txn, build->id, {}); notifyBuildFinished(txn, build->id, {});
txn.commit(); txn.commit();
build->finishedInDB = true; build->finishedInDB = true;
@ -255,7 +254,7 @@ bool State::getQueuedBuilds(Connection & conn,
{ {
auto mc = startDbUpdate(); auto mc = startDbUpdate();
pqxx::work txn(conn); pqxx::work txn(conn);
time_t now = time(nullptr); time_t now = time(0);
if (!buildOneDone && build->id == buildOne) buildOneDone = true; if (!buildOneDone && build->id == buildOne) buildOneDone = true;
printMsg(lvlInfo, "marking build %1% as succeeded (cached)", build->id); printMsg(lvlInfo, "marking build %1% as succeeded (cached)", build->id);
markSucceededBuild(txn, build, res, true, now, now); markSucceededBuild(txn, build, res, true, now, now);
@ -356,7 +355,7 @@ void State::processQueueChange(Connection & conn)
pqxx::work txn(conn); pqxx::work txn(conn);
auto res = txn.exec("select id, globalPriority from Builds where finished = 0"); auto res = txn.exec("select id, globalPriority from Builds where finished = 0");
for (auto const & row : res) for (auto const & row : res)
currentIds[row["id"].as<BuildID>()] = row["globalPriority"].as<int>(); currentIds[row["id"].as<BuildID>()] = row["globalPriority"].as<BuildID>();
} }
{ {
@ -439,7 +438,7 @@ Step::ptr State::createStep(ref<Store> destStore,
Build::ptr referringBuild, Step::ptr referringStep, std::set<StorePath> & finishedDrvs, Build::ptr referringBuild, Step::ptr referringStep, std::set<StorePath> & finishedDrvs,
std::set<Step::ptr> & newSteps, std::set<Step::ptr> & newRunnable) std::set<Step::ptr> & newSteps, std::set<Step::ptr> & newRunnable)
{ {
if (finishedDrvs.find(drvPath) != finishedDrvs.end()) return nullptr; if (finishedDrvs.find(drvPath) != finishedDrvs.end()) return 0;
/* Check if the requested step already exists. If not, create a /* Check if the requested step already exists. If not, create a
new step. In any case, make the step reachable from new step. In any case, make the step reachable from
@ -517,7 +516,7 @@ Step::ptr State::createStep(ref<Store> destStore,
std::map<DrvOutput, std::optional<StorePath>> paths; std::map<DrvOutput, std::optional<StorePath>> paths;
for (auto & [outputName, maybeOutputPath] : destStore->queryPartialDerivationOutputMap(drvPath, &*localStore)) { for (auto & [outputName, maybeOutputPath] : destStore->queryPartialDerivationOutputMap(drvPath, &*localStore)) {
auto outputHash = outputHashes.at(outputName); auto outputHash = outputHashes.at(outputName);
paths.insert({{.drvHash=outputHash, .outputName=outputName}, maybeOutputPath}); paths.insert({{outputHash, outputName}, maybeOutputPath});
} }
auto missing = getMissingRemotePaths(destStore, paths); auto missing = getMissingRemotePaths(destStore, paths);
@ -561,7 +560,7 @@ Step::ptr State::createStep(ref<Store> destStore,
auto & path = *pathOpt; auto & path = *pathOpt;
try { try {
time_t startTime = time(nullptr); time_t startTime = time(0);
if (localStore->isValidPath(path)) if (localStore->isValidPath(path))
printInfo("copying output ‘%1%’ of ‘%2%’ from local store", printInfo("copying output ‘%1%’ of ‘%2%’ from local store",
@ -579,7 +578,7 @@ Step::ptr State::createStep(ref<Store> destStore,
StorePathSet { path }, StorePathSet { path },
NoRepair, CheckSigs, NoSubstitute); NoRepair, CheckSigs, NoSubstitute);
time_t stopTime = time(nullptr); time_t stopTime = time(0);
{ {
auto mc = startDbUpdate(); auto mc = startDbUpdate();
@ -603,7 +602,7 @@ Step::ptr State::createStep(ref<Store> destStore,
// FIXME: check whether all outputs are in the binary cache. // FIXME: check whether all outputs are in the binary cache.
if (valid) { if (valid) {
finishedDrvs.insert(drvPath); finishedDrvs.insert(drvPath);
return nullptr; return 0;
} }
/* No, we need to build. */ /* No, we need to build. */
@ -611,7 +610,7 @@ Step::ptr State::createStep(ref<Store> destStore,
/* Create steps for the dependencies. */ /* Create steps for the dependencies. */
for (auto & i : step->drv->inputDrvs.map) { for (auto & i : step->drv->inputDrvs.map) {
auto dep = createStep(destStore, conn, build, i.first, nullptr, step, finishedDrvs, newSteps, newRunnable); auto dep = createStep(destStore, conn, build, i.first, 0, step, finishedDrvs, newSteps, newRunnable);
if (dep) { if (dep) {
auto step_(step->state.lock()); auto step_(step->state.lock());
step_->deps.insert(dep); step_->deps.insert(dep);
@ -659,11 +658,11 @@ Jobset::ptr State::createJobset(pqxx::work & txn,
auto res2 = txn.exec_params auto res2 = txn.exec_params
("select s.startTime, s.stopTime from BuildSteps s join Builds b on build = id " ("select s.startTime, s.stopTime from BuildSteps s join Builds b on build = id "
"where s.startTime is not null and s.stopTime > $1 and jobset_id = $2", "where s.startTime is not null and s.stopTime > $1 and jobset_id = $2",
time(nullptr) - Jobset::schedulingWindow * 10, time(0) - Jobset::schedulingWindow * 10,
jobsetID); jobsetID);
for (auto const & row : res2) { for (auto const & row : res2) {
auto startTime = row["startTime"].as<time_t>(); time_t startTime = row["startTime"].as<time_t>();
auto stopTime = row["stopTime"].as<time_t>(); time_t stopTime = row["stopTime"].as<time_t>();
jobset->addStep(startTime, stopTime - startTime); jobset->addStep(startTime, stopTime - startTime);
} }
@ -703,7 +702,7 @@ BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store>
"where finished = 1 and (buildStatus = 0 or buildStatus = 6) and path = $1", "where finished = 1 and (buildStatus = 0 or buildStatus = 6) and path = $1",
localStore->printStorePath(output)); localStore->printStorePath(output));
if (r.empty()) continue; if (r.empty()) continue;
auto id = r[0][0].as<BuildID>(); BuildID id = r[0][0].as<BuildID>();
printInfo("reusing build %d", id); printInfo("reusing build %d", id);

63
src/hydra-queue-runner/state.hh
View file

@ -8,7 +8,6 @@
#include <queue> #include <queue>
#include <regex> #include <regex>
#include <semaphore> #include <semaphore>
#include <utility>
#include <prometheus/counter.h> #include <prometheus/counter.h>
#include <prometheus/gauge.h> #include <prometheus/gauge.h>
@ -27,16 +26,16 @@
#include "machines.hh" #include "machines.hh"
using BuildID = unsigned int; typedef unsigned int BuildID;
using JobsetID = unsigned int; typedef unsigned int JobsetID;
using system_time = std::chrono::time_point<std::chrono::system_clock>; typedef std::chrono::time_point<std::chrono::system_clock> system_time;
using counter = std::atomic<unsigned long>; typedef std::atomic<unsigned long> counter;
enum BuildStatus { typedef enum {
bsSuccess = 0, bsSuccess = 0,
bsFailed = 1, bsFailed = 1,
bsDepFailed = 2, // builds only bsDepFailed = 2, // builds only
@ -50,10 +49,10 @@ enum BuildStatus {
bsNarSizeLimitExceeded = 11, bsNarSizeLimitExceeded = 11,
bsNotDeterministic = 12, bsNotDeterministic = 12,
bsBusy = 100, // not stored bsBusy = 100, // not stored
}; } BuildStatus;
enum StepState { typedef enum {
ssPreparing = 1, ssPreparing = 1,
ssConnecting = 10, ssConnecting = 10,
ssSendingInputs = 20, ssSendingInputs = 20,
@ -61,7 +60,7 @@ enum StepState {
ssWaitingForLocalSlot = 35, ssWaitingForLocalSlot = 35,
ssReceivingOutputs = 40, ssReceivingOutputs = 40,
ssPostProcessing = 50, ssPostProcessing = 50,
}; } StepState;
struct RemoteResult struct RemoteResult
@ -79,7 +78,7 @@ struct RemoteResult
unsigned int overhead = 0; unsigned int overhead = 0;
nix::Path logFile; nix::Path logFile;
[[nodiscard]] BuildStatus buildStatus() const BuildStatus buildStatus() const
{ {
return stepStatus == bsCachedFailure ? bsFailed : stepStatus; return stepStatus == bsCachedFailure ? bsFailed : stepStatus;
} }
@ -96,10 +95,10 @@ class Jobset
{ {
public: public:
using ptr = std::shared_ptr<Jobset>; typedef std::shared_ptr<Jobset> ptr;
using wptr = std::weak_ptr<Jobset>; typedef std::weak_ptr<Jobset> wptr;
static const time_t schedulingWindow = static_cast<time_t>(24 * 60 * 60); static const time_t schedulingWindow = 24 * 60 * 60;
private: private:
@ -116,7 +115,7 @@ public:
return (double) seconds / shares; return (double) seconds / shares;
} }
void setShares(unsigned int shares_) void setShares(int shares_)
{ {
assert(shares_ > 0); assert(shares_ > 0);
shares = shares_; shares = shares_;
@ -132,8 +131,8 @@ public:
struct Build struct Build
{ {
using ptr = std::shared_ptr<Build>; typedef std::shared_ptr<Build> ptr;
using wptr = std::weak_ptr<Build>; typedef std::weak_ptr<Build> wptr;
BuildID id; BuildID id;
nix::StorePath drvPath; nix::StorePath drvPath;
@ -164,8 +163,8 @@ struct Build
struct Step struct Step
{ {
using ptr = std::shared_ptr<Step>; typedef std::shared_ptr<Step> ptr;
using wptr = std::weak_ptr<Step>; typedef std::weak_ptr<Step> wptr;
nix::StorePath drvPath; nix::StorePath drvPath;
std::unique_ptr<nix::Derivation> drv; std::unique_ptr<nix::Derivation> drv;
@ -222,8 +221,13 @@ struct Step
nix::Sync<State> state; nix::Sync<State> state;
Step(nix::StorePath drvPath) : drvPath(std::move(drvPath)) Step(const nix::StorePath & drvPath) : drvPath(drvPath)
{ } { }
~Step()
{
//printMsg(lvlError, format("destroying step %1%") % drvPath);
}
}; };
@ -235,7 +239,7 @@ void visitDependencies(std::function<void(Step::ptr)> visitor, Step::ptr step);
struct Machine : nix::Machine struct Machine : nix::Machine
{ {
using ptr = std::shared_ptr<Machine>; typedef std::shared_ptr<Machine> ptr;
/* TODO Get rid of: `nix::Machine::storeUri` is normalized in a way /* TODO Get rid of: `nix::Machine::storeUri` is normalized in a way
we are not yet used to, but once we are, we don't need this. */ we are not yet used to, but once we are, we don't need this. */
@ -250,7 +254,7 @@ struct Machine : nix::Machine
float speedFactorFloat = 1.0; float speedFactorFloat = 1.0;
struct State { struct State {
using ptr = std::shared_ptr<State>; typedef std::shared_ptr<State> ptr;
counter currentJobs{0}; counter currentJobs{0};
counter nrStepsDone{0}; counter nrStepsDone{0};
counter totalStepTime{0}; // total time for steps, including closure copying counter totalStepTime{0}; // total time for steps, including closure copying
@ -354,22 +358,22 @@ private:
bool useSubstitutes = false; bool useSubstitutes = false;
/* The queued builds. */ /* The queued builds. */
using Builds = std::map<BuildID, Build::ptr>; typedef std::map<BuildID, Build::ptr> Builds;
nix::Sync<Builds> builds; nix::Sync<Builds> builds;
/* The jobsets. */ /* The jobsets. */
using Jobsets = std::map<std::pair<std::string, std::string>, Jobset::ptr>; typedef std::map<std::pair<std::string, std::string>, Jobset::ptr> Jobsets;
nix::Sync<Jobsets> jobsets; nix::Sync<Jobsets> jobsets;
/* All active or pending build steps (i.e. dependencies of the /* All active or pending build steps (i.e. dependencies of the
queued builds). Note that these are weak pointers. Steps are queued builds). Note that these are weak pointers. Steps are
kept alive by being reachable from Builds or by being in kept alive by being reachable from Builds or by being in
progress. */ progress. */
using Steps = std::map<nix::StorePath, Step::wptr>; typedef std::map<nix::StorePath, Step::wptr> Steps;
nix::Sync<Steps> steps; nix::Sync<Steps> steps;
/* Build steps that have no unbuilt dependencies. */ /* Build steps that have no unbuilt dependencies. */
using Runnable = std::list<Step::wptr>; typedef std::list<Step::wptr> Runnable;
nix::Sync<Runnable> runnable; nix::Sync<Runnable> runnable;
/* CV for waking up the dispatcher. */ /* CV for waking up the dispatcher. */
@ -381,7 +385,7 @@ private:
/* The build machines. */ /* The build machines. */
std::mutex machinesReadyLock; std::mutex machinesReadyLock;
using Machines = std::map<std::string, Machine::ptr>; typedef std::map<std::string, Machine::ptr> Machines;
nix::Sync<Machines> machines; // FIXME: use atomic_shared_ptr nix::Sync<Machines> machines; // FIXME: use atomic_shared_ptr
/* Throttler for CPU-bound local work. */ /* Throttler for CPU-bound local work. */
@ -397,7 +401,6 @@ private:
counter nrStepsDone{0}; counter nrStepsDone{0};
counter nrStepsBuilding{0}; counter nrStepsBuilding{0};
counter nrStepsCopyingTo{0}; counter nrStepsCopyingTo{0};
counter nrStepsWaitingForDownloadSlot{0};
counter nrStepsCopyingFrom{0}; counter nrStepsCopyingFrom{0};
counter nrStepsWaiting{0}; counter nrStepsWaiting{0};
counter nrUnsupportedSteps{0}; counter nrUnsupportedSteps{0};
@ -428,7 +431,7 @@ private:
struct MachineReservation struct MachineReservation
{ {
using ptr = std::shared_ptr<MachineReservation>; typedef std::shared_ptr<MachineReservation> ptr;
State & state; State & state;
Step::ptr step; Step::ptr step;
Machine::ptr machine; Machine::ptr machine;
@ -531,7 +534,7 @@ private:
void finishBuildStep(pqxx::work & txn, const RemoteResult & result, BuildID buildId, unsigned int stepNr, void finishBuildStep(pqxx::work & txn, const RemoteResult & result, BuildID buildId, unsigned int stepNr,
const std::string & machine); const std::string & machine);
unsigned int createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime, int createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime,
Build::ptr build, const nix::StorePath & drvPath, const nix::Derivation drv, const std::string & outputName, const nix::StorePath & storePath); Build::ptr build, const nix::StorePath & drvPath, const nix::Derivation drv, const std::string & outputName, const nix::StorePath & storePath);
void updateBuild(pqxx::work & txn, Build::ptr build, BuildStatus status); void updateBuild(pqxx::work & txn, Build::ptr build, BuildStatus status);
@ -620,6 +623,8 @@ private:
void addRoot(const nix::StorePath & storePath); void addRoot(const nix::StorePath & storePath);
void runMetricsExporter();
public: public:
void showStatus(); void showStatus();

24
src/lib/Hydra/Controller/API.pm
View file

@ -242,35 +242,23 @@ sub push : Chained('api') PathPart('push') Args(0) {
$c->{stash}->{json}->{jobsetsTriggered} = []; $c->{stash}->{json}->{jobsetsTriggered} = [];
my $force = exists $c->request->query_params->{force}; my $force = exists $c->request->query_params->{force};
my @jobsetNames = split /,/, ($c->request->query_params->{jobsets} // ""); my @jobsets = split /,/, ($c->request->query_params->{jobsets} // "");
my @jobsets; foreach my $s (@jobsets) {
foreach my $s (@jobsetNames) {
my ($p, $j) = parseJobsetName($s); my ($p, $j) = parseJobsetName($s);
my $jobset = $c->model('DB::Jobsets')->find($p, $j); my $jobset = $c->model('DB::Jobsets')->find($p, $j);
push @jobsets, $jobset if defined $jobset; next unless defined $jobset && ($force || ($jobset->project->enabled && $jobset->enabled));
triggerJobset($self, $c, $jobset, $force);
} }
my @repos = split /,/, ($c->request->query_params->{repos} // ""); my @repos = split /,/, ($c->request->query_params->{repos} // "");
foreach my $r (@repos) { foreach my $r (@repos) {
foreach ($c->model('DB::Jobsets')->search( triggerJobset($self, $c, $_, $force) foreach $c->model('DB::Jobsets')->search(
{ 'project.enabled' => 1, 'me.enabled' => 1 }, { 'project.enabled' => 1, 'me.enabled' => 1 },
{ {
join => 'project', join => 'project',
where => \ [ 'exists (select 1 from JobsetInputAlts where project = me.project and jobset = me.name and value = ?)', [ 'value', $r ] ], where => \ [ 'exists (select 1 from JobsetInputAlts where project = me.project and jobset = me.name and value = ?)', [ 'value', $r ] ],
order_by => 'me.id DESC' order_by => 'me.id DESC'
})) { });
push @jobsets, $_;
}
}
foreach my $jobset (@jobsets) {
requireRestartPrivileges($c, $jobset->project);
}
foreach my $jobset (@jobsets) {
next unless defined $jobset && ($force || ($jobset->project->enabled && $jobset->enabled));
triggerJobset($self, $c, $jobset, $force);
} }
$self->status_ok( $self->status_ok(

28
t/Hydra/Controller/API/checks.t
View file

@ -35,17 +35,6 @@ my $queuedBuilds = $ctx->makeAndEvaluateJobset(
build => 0 build => 0
); );
# Login and save cookie for future requests
my $req = request(POST '/login',
Referer => 'http://localhost/',
Content => {
username => 'root',
password => 'rootPassword'
}
);
is($req->code, 302, "Logging in gets a 302");
my $cookie = $req->header("set-cookie");
subtest "/api/queue" => sub { subtest "/api/queue" => sub {
my $response = request(GET '/api/queue?nr=1'); my $response = request(GET '/api/queue?nr=1');
ok($response->is_success, "The API enpdoint showing the queue returns 200."); ok($response->is_success, "The API enpdoint showing the queue returns 200.");
@ -113,17 +102,6 @@ subtest "/api/nrbuilds" => sub {
}; };
subtest "/api/push" => sub { subtest "/api/push" => sub {
subtest "without authentication" => sub {
my $build = $finishedBuilds->{"one_job"};
my $jobset = $build->jobset;
my $projectName = $jobset->project->name;
my $jobsetName = $jobset->name;
is($jobset->forceeval, undef, "The existing jobset is not set to be forced to eval");
my $response = request(GET "/api/push?jobsets=$projectName:$jobsetName&force=1");
is($response->code, 403, "The API enpdoint for triggering jobsets requires authentication.");
};
subtest "with a specific jobset" => sub { subtest "with a specific jobset" => sub {
my $build = $finishedBuilds->{"one_job"}; my $build = $finishedBuilds->{"one_job"};
my $jobset = $build->jobset; my $jobset = $build->jobset;
@ -131,8 +109,7 @@ subtest "/api/push" => sub {
my $jobsetName = $jobset->name; my $jobsetName = $jobset->name;
is($jobset->forceeval, undef, "The existing jobset is not set to be forced to eval"); is($jobset->forceeval, undef, "The existing jobset is not set to be forced to eval");
my $response = request(GET "/api/push?jobsets=$projectName:$jobsetName&force=1", my $response = request(GET "/api/push?jobsets=$projectName:$jobsetName&force=1");
Cookie => $cookie);
ok($response->is_success, "The API enpdoint for triggering jobsets returns 200."); ok($response->is_success, "The API enpdoint for triggering jobsets returns 200.");
my $data = is_json($response); my $data = is_json($response);
@ -151,8 +128,7 @@ subtest "/api/push" => sub {
print STDERR $repo; print STDERR $repo;
my $response = request(GET "/api/push?repos=$repo&force=1", my $response = request(GET "/api/push?repos=$repo&force=1");
Cookie => $cookie);
ok($response->is_success, "The API enpdoint for triggering jobsets returns 200."); ok($response->is_success, "The API enpdoint for triggering jobsets returns 200.");
my $data = is_json($response); my $data = is_json($response);

12
t/Hydra/Controller/Project/delete.t
View file

@ -11,14 +11,20 @@ my $ctx = test_context();
Catalyst::Test->import('Hydra'); Catalyst::Test->import('Hydra');
$ctx->db(); # Ensure DB initialization. my $user = $ctx->db()->resultset('Users')->create({
username => 'alice',
emailaddress => 'root@invalid.org',
password => '!'
});
$user->setPassword('foobar');
$user->userroles->update_or_create({ role => 'admin' });
# Login and save cookie for future requests # Login and save cookie for future requests
my $req = request(POST '/login', my $req = request(POST '/login',
Referer => 'http://localhost/', Referer => 'http://localhost/',
Content => { Content => {
username => 'root', username => 'alice',
password => 'rootPassword' password => 'foobar'
} }
); );
is($req->code, 302, "Logging in gets a 302"); is($req->code, 302, "Logging in gets a 302");

6
t/lib/HydraTestContext.pm
View file

@ -115,13 +115,11 @@ sub db {
$self->{_db} = Hydra::Model::DB->new(); $self->{_db} = Hydra::Model::DB->new();
if (!(defined $setup && $setup == 0)) { if (!(defined $setup && $setup == 0)) {
my $user = $self->{_db}->resultset('Users')->create({ $self->{_db}->resultset('Users')->create({
username => "root", username => "root",
emailaddress => 'root@invalid.org', emailaddress => 'root@invalid.org',
password => '!' password => ''
}); });
$user->setPassword('rootPassword');
$user->userroles->update_or_create({ role => 'admin' });
} }
} }