ahuston-0/hydra
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

Merge branch 'master' into libpqxx_undeprecate

Browse Source
This commit is contained in:
Kevin Quick
2020-04-01 11:54:41 -07:00
committed by GitHub
parent ec8b2970e0 2d092a6fbc
commit a055796ef5
88 changed files with 1667 additions and 1309 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/hydra-queue-runner/hydra-queue-runner.cc
View File

@ -39,14 +39,15 @@ static uint64_t getMemSize()
std::string getEnvOrDie(const std::string & key)
{
char * value = getenv(key.c_str());
auto value = getEnv(key);
if (!value) throw Error("environment variable '%s' is not set", key);
return value;
return *value;
}
State::State()
: config(std::make_unique<::Config>())
, maxUnsupportedTime(config->getIntOption("max_unsupported_time", 0))
, dbPool(config->getIntOption("max_db_connections", 128))
, memoryTokens(config->getIntOption("nar_buffer_size", getMemSize() / 2))
, maxOutputSize(config->getIntOption("max_output_size", 2ULL << 30))
@ -161,7 +162,7 @@ void State::monitorMachinesFile()
{
string defaultMachinesFile = "/etc/nix/machines";
auto machinesFiles = tokenizeString<std::vector<Path>>(
getEnv("NIX_REMOTE_SYSTEMS", pathExists(defaultMachinesFile) ? defaultMachinesFile : ""), ":");
getEnv("NIX_REMOTE_SYSTEMS").value_or(pathExists(defaultMachinesFile) ? defaultMachinesFile : ""), ":");
if (machinesFiles.empty()) {
parseMachines("localhost " +
@ -219,6 +220,7 @@ void State::monitorMachinesFile()
sleep(30);
} catch (std::exception & e) {
printMsg(lvlError, format("reloading machines file: %1%") % e.what());
sleep(5);
}
}
}
@ -253,7 +255,7 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
buildId,
stepNr,
0, // == build
step->drvPath,
localStore->printStorePath(step->drvPath),
status == bsBusy ? 1 : 0,
startTime != 0 ? std::make_optional(startTime) : std::nullopt,
step->drv.platform,
@ -268,7 +270,7 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
for (auto & output : step->drv.outputs)
txn.exec_params0
("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
buildId, stepNr, output.first, output.second.path);
buildId, stepNr, output.first, localStore->printStorePath(output.second.path));
if (status == bsBusy)
txn.exec(fmt("notify step_started, '%d\t%d'", buildId, stepNr));
@ -309,7 +311,7 @@ void State::finishBuildStep(pqxx::work & txn, const RemoteResult & result,
int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime,
Build::ptr build, const Path & drvPath, const string & outputName, const Path & storePath)
Build::ptr build, const StorePath & drvPath, const string & outputName, const StorePath & storePath)
{
restart:
auto stepNr = allocBuildStep(txn, build->id);
@ -319,7 +321,7 @@ int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t sto
build->id,
stepNr,
1, // == substitution
drvPath,
(localStore->printStorePath(drvPath)),
0,
0,
startTime,
@ -329,7 +331,8 @@ int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t sto
txn.exec_params0
("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
build->id, stepNr, outputName, storePath);
build->id, stepNr, outputName,
localStore->printStorePath(storePath));
return stepNr;
}
@ -450,7 +453,7 @@ bool State::checkCachedFailure(Step::ptr step, Connection & conn)
{
pqxx::work txn(conn);
for (auto & path : step->drv.outputPaths())
if (!txn.exec_params("select 1 from FailedPaths where path = $1", path).empty())
if (!txn.exec_params("select 1 from FailedPaths where path = $1", localStore->printStorePath(path)).empty())
return true;
return false;
}
@ -486,7 +489,7 @@ std::shared_ptr<PathLocks> State::acquireGlobalLock()
}
void State::dumpStatus(Connection & conn, bool log)
void State::dumpStatus(Connection & conn)
{
std::ostringstream out;
@ -518,6 +521,7 @@ void State::dumpStatus(Connection & conn, bool log)
root.attr("nrStepsCopyingTo", nrStepsCopyingTo);
root.attr("nrStepsCopyingFrom", nrStepsCopyingFrom);
root.attr("nrStepsWaiting", nrStepsWaiting);
root.attr("nrUnsupportedSteps", nrUnsupportedSteps);
root.attr("bytesSent", bytesSent);
root.attr("bytesReceived", bytesReceived);
root.attr("nrBuildsRead", nrBuildsRead);
@ -666,11 +670,6 @@ void State::dumpStatus(Connection & conn, bool log)
}
}
if (log && time(0) >= lastStatusLogged + statusLogInterval) {
printMsg(lvlInfo, format("status: %1%") % out.str());
lastStatusLogged = time(0);
}
{
auto mc = startDbUpdate();
pqxx::work txn(conn);
@ -762,7 +761,7 @@ void State::run(BuildID buildOne)
Store::Params localParams;
localParams["max-connections"] = "16";
localParams["max-connection-age"] = "600";
localStore = openStore(getEnv("NIX_REMOTE"), localParams);
localStore = openStore(getEnv("NIX_REMOTE").value_or(""), localParams);
auto storeUri = config->getStrOption("store_uri");
_destStore = storeUri == "" ? localStore : openStore(storeUri);
@ -779,7 +778,7 @@ void State::run(BuildID buildOne)
{
auto conn(dbPool.get());
clearBusy(*conn, 0);
dumpStatus(*conn, false);
dumpStatus(*conn);
}
std::thread(&State::monitorMachinesFile, this).detach();
@ -842,8 +841,8 @@ void State::run(BuildID buildOne)
auto conn(dbPool.get());
receiver dumpStatus_(*conn, "dump_status");
while (true) {
conn->await_notification(statusLogInterval / 2 + 1, 0);
dumpStatus(*conn, true);
conn->await_notification();
dumpStatus(*conn);
}
} catch (std::exception & e) {
printMsg(lvlError, format("main thread: %1%") % e.what());