From c0b9c8787f037ed32ab43604e7d84c2a814414cd Mon Sep 17 00:00:00 2001 From: Maximilian Bosch Date: Wed, 12 Mar 2025 10:40:07 +0100 Subject: [PATCH] Update flake and fix build MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In hydra-evaluator I would've done the same stuff with passing `aio` down to everything, but it seemed kinda weird to add it to every single method. IMHO, this is also a sign that there are things ongoing that don't need to be in the same struct, so I decided to split it off. Flake lock file updates: • Updated input 'lix': 'git+https://git.lix.systems/lix-project/lix?ref=refs/heads/main&rev=ca89e431a31527a014bfd0d529da2a8099027a5f' (2025-03-04) → 'git+https://git.lix.systems/lix-project/lix?ref=refs/heads/main&rev=2a336813ad2a4d64d027830507276da32927d215' (2025-03-16) • Updated input 'nixpkgs': 'github:NixOS/nixpkgs/6af28b834daca767a7ef99f8a7defa957d0ade6f' (2025-03-04) → 'github:NixOS/nixpkgs/a1185f4064c18a5db37c5c84e5638c78b46e3341' (2025-03-16) --- flake.lock | 14 +- src/hydra-evaluator/hydra-evaluator.cc | 352 ++++++++++--------- src/hydra-queue-runner/build-remote.cc | 10 +- src/hydra-queue-runner/build-result.cc | 4 +- src/hydra-queue-runner/builder.cc | 2 +- src/hydra-queue-runner/dispatcher.cc | 2 +- src/hydra-queue-runner/hydra-queue-runner.cc | 53 +-- src/hydra-queue-runner/queue-monitor.cc | 27 +- src/hydra-queue-runner/state.hh | 8 +- 9 files changed, 249 insertions(+), 223 deletions(-) diff --git a/flake.lock b/flake.lock index a98d73a53..db56e92b8 100644 --- a/flake.lock +++ b/flake.lock @@ -27,11 +27,11 @@ "pre-commit-hooks": "pre-commit-hooks" }, "locked": { - "lastModified": 1741082941, - "narHash": "sha256-mxMbmNSXLZ0G+4uPEXCodjRJffqh/Jq4X5pgFuQFZB0=", + "lastModified": 1742165831, + "narHash": "sha256-/ssmsf50UERQNlOcUuyfvigcnCNckAhAPZMo0+Y3mdY=", "ref": "refs/heads/main", - "rev": "ca89e431a31527a014bfd0d529da2a8099027a5f", - "revCount": 17577, + "rev": "2a336813ad2a4d64d027830507276da32927d215", + "revCount": 17664, "type": "git", "url": "https://git.lix.systems/lix-project/lix" }, @@ -58,11 +58,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1741048562, - "narHash": "sha256-W4YZ3fvWZiFYYyd900kh8P8wU6DHSiwaH0j4+fai1Sk=", + "lastModified": 1742136038, + "narHash": "sha256-DDe16FJk18sadknQKKG/9FbwEro7A57tg9vB5kxZ8kY=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "6af28b834daca767a7ef99f8a7defa957d0ade6f", + "rev": "a1185f4064c18a5db37c5c84e5638c78b46e3341", "type": "github" }, "original": { diff --git a/src/hydra-evaluator/hydra-evaluator.cc b/src/hydra-evaluator/hydra-evaluator.cc index b03ba8bfe..711fdb8da 100644 --- a/src/hydra-evaluator/hydra-evaluator.cc +++ b/src/hydra-evaluator/hydra-evaluator.cc @@ -79,48 +79,162 @@ enum class EvaluationStyle ONE_AT_A_TIME = 3, }; -struct Evaluator +struct Jobset { - std::unique_ptr config; + JobsetId name; + std::optional evaluation_style; + time_t lastCheckedTime, triggerTime; + time_t checkInterval; + Pid pid; +}; - nix::Pool dbPool; +using Jobsets = std::map; - struct Jobset - { - JobsetId name; - std::optional evaluation_style; - time_t lastCheckedTime, triggerTime; - time_t checkInterval; - Pid pid; - }; - - using Jobsets = std::map; - - std::optional evalOne; - - const size_t maxEvals; - - struct State - { - size_t runningEvals = 0; - Jobsets jobsets; - }; +struct State +{ + size_t runningEvals = 0; + Jobsets jobsets; +}; +struct EvaluatorState +{ Sync state_; std::condition_variable childStarted; std::condition_variable maybeDoWork; - const time_t notTriggered = std::numeric_limits::max(); + std::optional evalOne; - Evaluator() - : config(std::make_unique()) - , maxEvals(std::max((size_t) 1, (size_t) config->getIntOption("max_concurrent_evals", 4))) - { } + const time_t notTriggered = std::numeric_limits::max(); +}; + +static void reaper(EvaluatorState & evalState, nix::Pool & dbPool) +{ + AsyncIoRoot aio; + + while (true) { + { + auto state(evalState.state_.lock()); + while (!state->runningEvals) + state.wait(evalState.childStarted); + } + + int status; + pid_t pid = waitpid(-1, &status, 0); + if (pid == -1) { + if (errno == EINTR) continue; + throw SysError("waiting for children"); + } + + { + auto state(evalState.state_.lock()); + assert(state->runningEvals); + state->runningEvals--; + + // FIXME: should use a map. + for (auto & i : state->jobsets) { + auto & jobset(i.second); + + if (jobset.pid.get() == pid) { + printInfo("evaluation of jobset ‘%s’ %s", + jobset.name.display(), statusToString(status)); + + auto now = time(nullptr); + + jobset.triggerTime = evalState.notTriggered; + jobset.lastCheckedTime = now; + + try { + + auto conn(aio.blockOn(dbPool.get())); + pqxx::work txn(*conn); + + /* Clear the trigger time to prevent this + jobset from getting stuck in an endless + failing eval loop. */ + txn.exec_params0 + ("update Jobsets set triggerTime = null where id = $1 and startTime is not null and triggerTime <= startTime", + jobset.name.id); + + /* Clear the start time. */ + txn.exec_params0 + ("update Jobsets set startTime = null where id = $1", + jobset.name.id); + + if (!WIFEXITED(status) || WEXITSTATUS(status) > 1) { + txn.exec_params0 + ("update Jobsets set errorMsg = $1, lastCheckedTime = $2, errorTime = $2, fetchErrorMsg = null where id = $3", + fmt("evaluation %s", statusToString(status)), + now, + jobset.name.id); + } + + txn.commit(); + + } catch (std::exception & e) { + printError("exception setting jobset error: %s", e.what()); + } + + jobset.pid.release(); + evalState.maybeDoWork.notify_one(); + + if (evalState.evalOne) std::_Exit(0); + + break; + } + } + } + } +} + +struct Monitor +{ + nix::Pool & dbPool; + EvaluatorState & evalState; + nix::AsyncIoRoot & aio; + + Monitor(nix::Pool & dbPool, EvaluatorState & state, nix::AsyncIoRoot & aio) + : dbPool(dbPool) + , evalState(state) + , aio(aio) + {} + + /* 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(aio.blockOn(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(); + evalState.maybeDoWork.notify_one(); + conn->await_notification(); + printInfo("received jobset event"); + } + + } catch (pqxx::broken_connection & e) { + printError("Database connection broken: %s", e.what()); + std::_Exit(1); + } catch (std::exception & e) { + printError("exception in database monitor thread: %s", e.what()); + sleep(30); + } + } + } void readJobsets() { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); pqxx::work txn(*conn); @@ -131,20 +245,20 @@ struct Evaluator "where j.enabled != 0 and p.enabled != 0"); - auto state(state_.lock()); + auto state(evalState.state_.lock()); std::set seen; for (auto const & row : res) { auto name = JobsetId{row["project"].as(), row["name"].as(), row["id"].as()}; - if (evalOne && name != *evalOne) continue; + if (evalState.evalOne && name != *evalState.evalOne) continue; auto res = state->jobsets.try_emplace(name, Jobset{.name=name}); auto & jobset = res.first->second; jobset.lastCheckedTime = row["lastCheckedTime"].as(0); - jobset.triggerTime = row["triggerTime"].as(notTriggered); + jobset.triggerTime = row["triggerTime"].as(evalState.notTriggered); jobset.checkInterval = row["checkInterval"].as(); int eval_style = row["jobset_enabled"].as(0); @@ -166,7 +280,7 @@ struct Evaluator seen.insert(name); } - if (evalOne && seen.empty()) { + if (evalState.evalOne && seen.empty()) { printError("the specified jobset does not exist or is disabled"); std::_Exit(1); } @@ -179,6 +293,25 @@ struct Evaluator i = state->jobsets.erase(i); } } +}; + +struct Evaluator +{ + std::unique_ptr config; + nix::Pool & dbPool; + EvaluatorState & evalState; + nix::AsyncIoRoot & aio; + const size_t maxEvals; + const time_t notTriggered = std::numeric_limits::max(); + + Evaluator(nix::Pool & dbPool, EvaluatorState & evalState, nix::AsyncIoRoot & aio) + : config(std::make_unique()) + , dbPool(dbPool) + , evalState(evalState) + , aio(aio) + , maxEvals(std::max((size_t) 1, (size_t) config->getIntOption("max_concurrent_evals", 4))) + { } + void startEval(State & state, Jobset & jobset) { @@ -189,7 +322,7 @@ struct Evaluator now - jobset.lastCheckedTime); { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); pqxx::work txn(*conn); txn.exec_params0 ("update Jobsets set startTime = $1 where id = $2", @@ -208,7 +341,7 @@ struct Evaluator state.runningEvals++; - childStarted.notify_one(); + evalState.childStarted.notify_one(); } bool shouldEvaluate(Jobset & jobset) @@ -240,7 +373,7 @@ struct Evaluator // is a ONE_AT_A_TIME jobset, ensure the previous jobset // has no remaining, unfinished work. - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); pqxx::work txn(*conn); @@ -302,7 +435,7 @@ struct Evaluator /* 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 (evalOne || + if (evalState.evalOne || (i->second.evaluation_style && shouldEvaluate(i->second))) sorted.push_back(i); @@ -325,9 +458,9 @@ struct Evaluator } } - void loop() + void loop(AsyncIoRoot & aio) { - auto state(state_.lock()); + auto state(evalState.state_.lock()); while (true) { @@ -345,149 +478,39 @@ struct Evaluator debug("waiting for %d s", sleepTime.count()); if (sleepTime == std::chrono::seconds::max()) - state.wait(maybeDoWork); + state.wait(evalState.maybeDoWork); else - state.wait_for(maybeDoWork, sleepTime); + state.wait_for(evalState.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 (pqxx::broken_connection & e) { - printError("Database connection broken: %s", e.what()); - std::_Exit(1); - } 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--; - - // FIXME: should use a map. - for (auto & i : state->jobsets) { - auto & jobset(i.second); - - if (jobset.pid.get() == pid) { - printInfo("evaluation of jobset ‘%s’ %s", - jobset.name.display(), statusToString(status)); - - auto now = time(nullptr); - - jobset.triggerTime = notTriggered; - jobset.lastCheckedTime = now; - - try { - - auto conn(dbPool.get()); - pqxx::work txn(*conn); - - /* Clear the trigger time to prevent this - jobset from getting stuck in an endless - failing eval loop. */ - txn.exec_params0 - ("update Jobsets set triggerTime = null where id = $1 and startTime is not null and triggerTime <= startTime", - jobset.name.id); - - /* Clear the start time. */ - txn.exec_params0 - ("update Jobsets set startTime = null where id = $1", - jobset.name.id); - - if (!WIFEXITED(status) || WEXITSTATUS(status) > 1) { - txn.exec_params0 - ("update Jobsets set errorMsg = $1, lastCheckedTime = $2, errorTime = $2, fetchErrorMsg = null where id = $3", - fmt("evaluation %s", statusToString(status)), - now, - jobset.name.id); - } - - txn.commit(); - - } catch (std::exception & e) { - printError("exception setting jobset error: %s", e.what()); - } - - jobset.pid.release(); - maybeDoWork.notify_one(); - - if (evalOne) std::_Exit(0); - - break; - } - } - } - } - } - void unlock() { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); pqxx::work txn(*conn); txn.exec("update Jobsets set startTime = null"); txn.commit(); } - void run() + void run(AsyncIoRoot & aio) { unlock(); /* Can't be bothered to shut down cleanly. Goodbye! */ auto callback = createInterruptCallback([&]() { std::_Exit(1); }); - std::thread reaperThread([&]() { reaper(); }); - - std::thread monitorThread([&]() { databaseMonitor(); }); + std::thread reaperThread([&]() { reaper(evalState, dbPool); }); + std::thread monitorThread([&]() { + AsyncIoRoot aio; + Monitor monitor(dbPool, evalState, aio); + monitor.databaseMonitor(); + }); while (true) { try { - loop(); + loop(aio); } catch (pqxx::broken_connection & e) { printError("Database connection broken: %s", e.what()); std::_Exit(1); @@ -510,8 +533,6 @@ int main(int argc, char * * argv) bool unlock = false; - Evaluator evaluator; - std::vector args; nix::AsyncIoRoot aio; @@ -524,15 +545,18 @@ int main(int argc, char * * argv) return true; }).parseCmdline(Strings(argv + 1, argv + argc)); + EvaluatorState evalState; + nix::Pool dbPool; + Evaluator evaluator(dbPool, evalState, aio); if (unlock) evaluator.unlock(); else { if (!args.empty()) { if (args.size() != 2) throw UsageError("Syntax: hydra-evaluator [ ]"); - evaluator.evalOne = JobsetName(args[0], args[1]); + evalState.evalOne = JobsetName(args[0], args[1]); } - evaluator.run(); + evaluator.run(aio); } }); } diff --git a/src/hydra-queue-runner/build-remote.cc b/src/hydra-queue-runner/build-remote.cc index 4fe4e2baa..b2cfca1a7 100644 --- a/src/hydra-queue-runner/build-remote.cc +++ b/src/hydra-queue-runner/build-remote.cc @@ -286,6 +286,7 @@ static BasicDerivation sendInputs( } static BuildResult performBuild( + AsyncIoRoot & aio, ::Machine::Connection & conn, Store & localStore, StorePath drvPath, @@ -337,7 +338,7 @@ static BuildResult performBuild( DerivationOutputsAndOptPaths drvOutputs = drv.outputsAndOptPaths(localStore); // Since this a `BasicDerivation`, `staticOutputHashes` will not // do any real work. - auto outputHashes = staticOutputHashes(localStore, drv); + auto outputHashes = aio.blockOn(staticOutputHashes(localStore, drv)); for (auto & [outputName, output] : drvOutputs) { auto outputPath = output.second; // We’ve just asserted that the output paths of the derivation @@ -606,6 +607,7 @@ void State::buildRemote(AsyncIoRoot & aio, updateStep(ssBuilding); BuildResult buildResult = build_remote::performBuild( + aio, conn, *localStore, step->drvPath, @@ -679,16 +681,16 @@ void State::buildRemote(AsyncIoRoot & aio, /* Register the outputs of the newly built drv */ if (experimentalFeatureSettings.isEnabled(Xp::CaDerivations)) { - auto outputHashes = staticOutputHashes(*localStore, *step->drv); + auto outputHashes = aio.blockOn(staticOutputHashes(*localStore, *step->drv)); for (auto & [outputName, realisation] : buildResult.builtOutputs) { // Register the resolved drv output - destStore->registerDrvOutput(realisation); + aio.blockOn(destStore->registerDrvOutput(realisation)); // Also register the unresolved one auto unresolvedRealisation = realisation; unresolvedRealisation.signatures.clear(); unresolvedRealisation.id.drvHash = outputHashes.at(outputName); - destStore->registerDrvOutput(unresolvedRealisation); + aio.blockOn(destStore->registerDrvOutput(unresolvedRealisation)); } } diff --git a/src/hydra-queue-runner/build-result.cc b/src/hydra-queue-runner/build-result.cc index 051784471..4df1f864b 100644 --- a/src/hydra-queue-runner/build-result.cc +++ b/src/hydra-queue-runner/build-result.cc @@ -25,7 +25,7 @@ BuildOutput getBuildOutput( res.outputs.insert({outputName, outputPath}); } for (auto & path : closure) { - auto info = store->queryPathInfo(path); + auto info = aio.blockOn(store->queryPathInfo(path)); res.closureSize += info->narSize; if (outputs.count(path)) res.size += info->narSize; } @@ -36,7 +36,7 @@ BuildOutput getBuildOutput( auto outputS = store->printStorePath(output); if (!narMembers.count(outputS)) { printInfo("fetching NAR contents of '%s'...", outputS); - auto source = store->narFromPath(output); + auto source = aio.blockOn(store->narFromPath(output)); extractNarData(source, outputS, narMembers); } } diff --git a/src/hydra-queue-runner/builder.cc b/src/hydra-queue-runner/builder.cc index a4a814e8f..6243f315f 100644 --- a/src/hydra-queue-runner/builder.cc +++ b/src/hydra-queue-runner/builder.cc @@ -38,7 +38,7 @@ void State::builder(MachineReservation::ptr reservation) activeSteps_.lock()->erase(activeStep); }); - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); try { auto destStore = getDestStore(); diff --git a/src/hydra-queue-runner/dispatcher.cc b/src/hydra-queue-runner/dispatcher.cc index f744d3ba2..d1b605b1d 100644 --- a/src/hydra-queue-runner/dispatcher.cc +++ b/src/hydra-queue-runner/dispatcher.cc @@ -378,7 +378,7 @@ void State::abortUnsupported(AsyncIoRoot & aio) aborted.insert(step); - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); std::set dependents; std::set steps; diff --git a/src/hydra-queue-runner/hydra-queue-runner.cc b/src/hydra-queue-runner/hydra-queue-runner.cc index c59b0216d..b71655391 100644 --- a/src/hydra-queue-runner/hydra-queue-runner.cc +++ b/src/hydra-queue-runner/hydra-queue-runner.cc @@ -599,7 +599,7 @@ std::optional State::acquireGlobalLock() } -void State::dumpStatus(Connection & conn) +void State::dumpStatus(Connection & conn, AsyncIoRoot & aio) { time_t now = time(nullptr); json statusJson = { @@ -715,21 +715,21 @@ void State::dumpStatus(Connection & conn) auto store = getDestStore(); - auto & stats = store->getStats(); + auto stats = aio.blockOn(store->getStats()); statusJson["store"] = { - {"narInfoRead", stats.narInfoRead.load()}, - {"narInfoReadAverted", stats.narInfoReadAverted.load()}, - {"narInfoMissing", stats.narInfoMissing.load()}, - {"narInfoWrite", stats.narInfoWrite.load()}, - {"narInfoCacheSize", stats.pathInfoCacheSize.load()}, - {"narRead", stats.narRead.load()}, - {"narReadBytes", stats.narReadBytes.load()}, - {"narReadCompressedBytes", stats.narReadCompressedBytes.load()}, - {"narWrite", stats.narWrite.load()}, - {"narWriteAverted", stats.narWriteAverted.load()}, - {"narWriteBytes", stats.narWriteBytes.load()}, - {"narWriteCompressedBytes", stats.narWriteCompressedBytes.load()}, - {"narWriteCompressionTimeMs", stats.narWriteCompressionTimeMs.load()}, + {"narInfoRead", stats.narInfoRead}, + {"narInfoReadAverted", stats.narInfoReadAverted}, + {"narInfoMissing", stats.narInfoMissing}, + {"narInfoWrite", stats.narInfoWrite}, + {"narInfoCacheSize", stats.pathInfoCacheSize}, + {"narRead", stats.narRead}, + {"narReadBytes", stats.narReadBytes}, + {"narReadCompressedBytes", stats.narReadCompressedBytes}, + {"narWrite", stats.narWrite}, + {"narWriteAverted", stats.narWriteAverted}, + {"narWriteBytes", stats.narWriteBytes}, + {"narWriteCompressedBytes", stats.narWriteCompressedBytes}, + {"narWriteCompressionTimeMs", stats.narWriteCompressionTimeMs}, {"narCompressionSavings", stats.narWriteBytes ? 1.0 - (double) stats.narWriteCompressedBytes / (double) stats.narWriteBytes @@ -779,9 +779,9 @@ void State::dumpStatus(Connection & conn) } -void State::showStatus() +void State::showStatus(AsyncIoRoot & aio) { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); receiver statusDumped(*conn, "status_dumped"); std::string status; @@ -825,13 +825,13 @@ void State::showStatus() } -void State::unlock() +void State::unlock(AsyncIoRoot & aio) { auto lock = acquireGlobalLock(); if (!lock) throw Error("hydra-queue-runner is currently running"); - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); clearBusy(*conn, 0); @@ -885,9 +885,9 @@ void State::run(AsyncIoRoot & aio, BuildID buildOne) } { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); clearBusy(*conn, 0); - dumpStatus(*conn); + dumpStatus(*conn, aio); } machinesReadyLock.lock(); @@ -899,6 +899,7 @@ void State::run(AsyncIoRoot & aio, BuildID buildOne) /* Periodically clean up orphaned busy steps in the database. */ std::thread([&]() { + AsyncIoRoot aio; while (true) { sleep(180); @@ -911,7 +912,7 @@ void State::run(AsyncIoRoot & aio, BuildID buildOne) } try { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); pqxx::work txn(*conn); for (auto & step : steps) { printMsg(lvlError, "cleaning orphaned step %d of build %d", step.second, step.first); @@ -934,12 +935,12 @@ void State::run(AsyncIoRoot & aio, BuildID buildOne) ‘hydra-queue-runner --status’). */ while (true) { try { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); try { receiver dumpStatus_(*conn, "dump_status"); while (true) { conn->await_notification(); - dumpStatus(*conn); + dumpStatus(*conn, aio); } } catch (pqxx::broken_connection & connEx) { printMsg(lvlError, "main thread: %s", connEx.what()); @@ -995,9 +996,9 @@ int main(int argc, char * * argv) State state{metricsAddrOpt}; if (status) - state.showStatus(); + state.showStatus(aio); else if (unlock) - state.unlock(); + state.unlock(aio); else state.run(aio, buildOne); }); diff --git a/src/hydra-queue-runner/queue-monitor.cc b/src/hydra-queue-runner/queue-monitor.cc index 04900a2db..ea8e7dbe1 100644 --- a/src/hydra-queue-runner/queue-monitor.cc +++ b/src/hydra-queue-runner/queue-monitor.cc @@ -13,10 +13,11 @@ using namespace nix; void State::queueMonitor() { + AsyncIoRoot aio; while (true) { - auto conn(dbPool.get()); + auto conn(aio.blockOn(dbPool.get())); try { - queueMonitorLoop(*conn); + queueMonitorLoop(aio, *conn); } catch (pqxx::broken_connection & e) { printMsg(lvlError, "queue monitor: %s", e.what()); printMsg(lvlError, "queue monitor: Reconnecting in 10s"); @@ -30,7 +31,7 @@ void State::queueMonitor() } -void State::queueMonitorLoop(Connection & conn) +void State::queueMonitorLoop(AsyncIoRoot & aio, Connection & conn) { receiver buildsAdded(conn, "builds_added"); receiver buildsRestarted(conn, "builds_restarted"); @@ -41,13 +42,11 @@ void State::queueMonitorLoop(Connection & conn) auto destStore = getDestStore(); - AsyncIoRoot aio; - bool quit = false; while (!quit) { auto t_before_work = std::chrono::steady_clock::now(); - localStore->clearPathInfoCache(); + localStore->clearPathInfoCache().wait(aio.kj.waitScope); bool done = getQueuedBuilds(aio, conn, destStore); @@ -156,7 +155,7 @@ bool State::getQueuedBuilds(AsyncIoRoot & aio, Connection & conn, nrAdded++; newBuildsByID.erase(build->id); - if (!localStore->isValidPath(build->drvPath)) { + if (!aio.blockOn(localStore->isValidPath(build->drvPath))) { /* Derivation has been GC'ed prematurely. */ printError("aborting GC'ed build %1%", build->id); if (!build->finishedInDB) { @@ -421,8 +420,8 @@ std::map> State::getMissingRemotePaths( auto missing(missing_.lock()); missing->insert({output, maybeOutputPath}); } else { - tp.enqueue([&] { - if (!destStore->isValidPath(*maybeOutputPath)) { + tp.enqueueWithAio([&](AsyncIoRoot & aio) { + if (!aio.blockOn(destStore->isValidPath(*maybeOutputPath))) { auto missing(missing_.lock()); missing->insert({output, maybeOutputPath}); } @@ -494,7 +493,7 @@ Step::ptr State::createStep(AsyncIoRoot & aio, ref destStore, ‘steps’ before this point, but that doesn't matter because it's not runnable yet, and other threads won't make it runnable while step->created == false. */ - step->drv = std::make_unique(localStore->readDerivation(drvPath)); + step->drv = std::make_unique(aio.blockOn(localStore->readDerivation(drvPath))); step->parsedDrv = std::make_unique(drvPath, *step->drv); step->preferLocalBuild = step->parsedDrv->willBuildLocally(*localStore); @@ -516,7 +515,7 @@ Step::ptr State::createStep(AsyncIoRoot & aio, ref destStore, throw PreviousFailure{step}; /* Are all outputs valid? */ - auto outputHashes = staticOutputHashes(*localStore, *(step->drv)); + auto outputHashes = aio.blockOn(staticOutputHashes(*localStore, *(step->drv))); std::map> paths; for (auto & [outputName, maybeOutputPath] : aio.blockOn(destStore->queryPartialDerivationOutputMap(drvPath, &*localStore))) { auto outputHash = outputHashes.at(outputName); @@ -535,7 +534,7 @@ Step::ptr State::createStep(AsyncIoRoot & aio, ref destStore, // If we don't know the output path from the destination // store, see if the local store can tell us. if (/* localStore != destStore && */ !pathOpt && experimentalFeatureSettings.isEnabled(Xp::CaDerivations)) - if (auto maybeRealisation = localStore->queryRealisation(i)) + if (auto maybeRealisation = aio.blockOn(localStore->queryRealisation(i))) pathOpt = maybeRealisation->outPath; if (!pathOpt) { @@ -545,7 +544,7 @@ Step::ptr State::createStep(AsyncIoRoot & aio, ref destStore, } auto & path = *pathOpt; - if (/* localStore != destStore && */ localStore->isValidPath(path)) + if (/* localStore != destStore && */ aio.blockOn(localStore->isValidPath(path))) avail++; else if (useSubstitutes) { SubstitutablePathInfos infos; @@ -566,7 +565,7 @@ Step::ptr State::createStep(AsyncIoRoot & aio, ref destStore, try { time_t startTime = time(nullptr); - if (localStore->isValidPath(path)) + if (aio.blockOn(localStore->isValidPath(path))) printInfo("copying output ‘%1%’ of ‘%2%’ from local store", localStore->printStorePath(path), localStore->printStorePath(drvPath)); diff --git a/src/hydra-queue-runner/state.hh b/src/hydra-queue-runner/state.hh index 16f2f2ca6..94337948c 100644 --- a/src/hydra-queue-runner/state.hh +++ b/src/hydra-queue-runner/state.hh @@ -539,7 +539,7 @@ private: void queueMonitor(); - void queueMonitorLoop(Connection & conn); + void queueMonitorLoop(nix::AsyncIoRoot & aio, Connection & conn); /* Check the queue for new builds. */ bool getQueuedBuilds(nix::AsyncIoRoot & aio, Connection & conn, nix::ref destStore); @@ -620,15 +620,15 @@ private: has it. */ std::optional acquireGlobalLock(); - void dumpStatus(Connection & conn); + void dumpStatus(Connection & conn, nix::AsyncIoRoot & aio); void addRoot(const nix::StorePath & storePath); public: - void showStatus(); + void showStatus(nix::AsyncIoRoot & aio); - void unlock(); + void unlock(nix::AsyncIoRoot & aio); void run(nix::AsyncIoRoot & aio, BuildID buildOne = 0); };