guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update flake and fix build

Browse Source 
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)
This commit is contained in:
Maximilian Bosch
2025-03-17 20:05:57 +00:00
parent 4b826e2255
commit c0b9c8787f
9 changed files with 249 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flake.lock
Generated
+7 -7
View File
@@ -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": {
src/hydra-evaluator/hydra-evaluator.cc
+188 -164
View File
@@ -79,48 +79,162 @@ enum class EvaluationStyle
ONE_AT_A_TIME = 3,
};
struct Evaluator
struct Jobset
{
std::unique_ptr<HydraConfig> config;
JobsetId name;
std::optional<EvaluationStyle> evaluation_style;
time_t lastCheckedTime, triggerTime;
time_t checkInterval;
Pid pid;
};
nix::Pool<Connection> dbPool;
using Jobsets = std::map<JobsetId, Jobset>;
struct Jobset
{
JobsetId name;
std::optional<EvaluationStyle> evaluation_style;
time_t lastCheckedTime, triggerTime;
time_t checkInterval;
Pid pid;
};
using Jobsets = std::map<JobsetId, Jobset>;
std::optional<JobsetName> 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> state_;
std::condition_variable childStarted;
std::condition_variable maybeDoWork;
const time_t notTriggered = std::numeric_limits<time_t>::max();
std::optional<JobsetName> evalOne;
Evaluator()
: config(std::make_unique<HydraConfig>())
, maxEvals(std::max((size_t) 1, (size_t) config->getIntOption("max_concurrent_evals", 4)))
{ }
const time_t notTriggered = std::numeric_limits<time_t>::max();
};
static void reaper(EvaluatorState & evalState, nix::Pool<Connection> & 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<Connection> & dbPool;
EvaluatorState & evalState;
nix::AsyncIoRoot & aio;
Monitor(nix::Pool<Connection> & 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<JobsetId> seen;
for (auto const & row : res) {
auto name = JobsetId{row["project"].as<std::string>(), row["name"].as<std::string>(), row["id"].as<int>()};
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<time_t>(0);
jobset.triggerTime = row["triggerTime"].as<time_t>(notTriggered);
jobset.triggerTime = row["triggerTime"].as<time_t>(evalState.notTriggered);
jobset.checkInterval = row["checkInterval"].as<time_t>();
int eval_style = row["jobset_enabled"].as<int>(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<HydraConfig> config;
nix::Pool<Connection> & dbPool;
EvaluatorState & evalState;
nix::AsyncIoRoot & aio;
const size_t maxEvals;
const time_t notTriggered = std::numeric_limits<time_t>::max();
Evaluator(nix::Pool<Connection> & dbPool, EvaluatorState & evalState, nix::AsyncIoRoot & aio)
: config(std::make_unique<HydraConfig>())
, 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<std::string> 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<Connection> dbPool;
Evaluator evaluator(dbPool, evalState, aio);
if (unlock)
evaluator.unlock();
else {
if (!args.empty()) {
if (args.size() != 2) throw UsageError("Syntax: hydra-evaluator [<project> <jobset>]");
evaluator.evalOne = JobsetName(args[0], args[1]);
evalState.evalOne = JobsetName(args[0], args[1]);
}
evaluator.run();
evaluator.run(aio);
}
});
}
src/hydra-queue-runner/build-remote.cc
+6 -4
View File
@@ -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));
}
}
src/hydra-queue-runner/build-result.cc
+2 -2
View File
@@ -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);
}
}
src/hydra-queue-runner/builder.cc
+1 -1
View File
@@ -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();
src/hydra-queue-runner/dispatcher.cc
+1 -1
View File
@@ -378,7 +378,7 @@ void State::abortUnsupported(AsyncIoRoot & aio)
aborted.insert(step);
auto conn(dbPool.get());
auto conn(aio.blockOn(dbPool.get()));
std::set<Build::ptr> dependents;
std::set<Step::ptr> steps;
src/hydra-queue-runner/hydra-queue-runner.cc
+27 -26
View File
@@ -599,7 +599,7 @@ std::optional<PathLock> 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);
});
src/hydra-queue-runner/queue-monitor.cc
+13 -14
View File
@@ -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<DrvOutput, std::optional<StorePath>> 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<Store> 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<Derivation>(localStore->readDerivation(drvPath));
step->drv = std::make_unique<Derivation>(aio.blockOn(localStore->readDerivation(drvPath)));
step->parsedDrv = std::make_unique<ParsedDerivation>(drvPath, *step->drv);
step->preferLocalBuild = step->parsedDrv->willBuildLocally(*localStore);
@@ -516,7 +515,7 @@ Step::ptr State::createStep(AsyncIoRoot & aio, ref<Store> destStore,
throw PreviousFailure{step};
/* Are all outputs valid? */
auto outputHashes = staticOutputHashes(*localStore, *(step->drv));
auto outputHashes = aio.blockOn(staticOutputHashes(*localStore, *(step->drv)));
std::map<DrvOutput, std::optional<StorePath>> 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<Store> 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<Store> 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<Store> 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));
src/hydra-queue-runner/state.hh
+4 -4
View File
@@ -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<nix::Store> destStore);
@@ -620,15 +620,15 @@ private:
has it. */
std::optional<nix::PathLock> 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);
};