hydra-eval-jobs: Parallelize

Dependencies

• [2] NQPGIRXX Revert "hydra-eval-jobs -> nix eval-hydra-jobs"

Change contents

• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 3
  [2.566]→[2.566:623](∅→∅)
  −

  − #define GC_LINUX_THREADS 1

  − #include <gc/gc_allocator.h>
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 9
  [2.733]→[2.733:752](∅→∅)
  − #include "json.hh"
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 14
  [2.883]

  [2.883]
  + #include "attr-path.hh"

  + #include "derivations.hh"
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 21
  [2.957]

  [2.957]
  + #include <sys/resource.h>

  +

  + #include <nlohmann/json.hpp>
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 26
  [2.979]

  [2.979]
  +

  + static Path gcRootsDir;

  + static size_t maxMemorySize;

  +

  + struct MyArgs : MixEvalArgs, MixCommonArgs

  + {

  + Path releaseExpr;

  + bool flake = false;

  + bool dryRun = false;
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 36
  [2.980]

  [2.980]
  + MyArgs() : MixCommonArgs("hydra-eval-jobs")

  + {

  + mkFlag()

  + .longName("help")

  + .description("show usage information")

  + .handler([&]() {

  + printHelp(programName, std::cout);

  + throw Exit();

  + });
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 46
  [2.981]→[2.981:1005](∅→∅)
  − static Path gcRootsDir;
  [2.981]

  [2.1005]
  + mkFlag()

  + .longName("gc-roots-dir")

  + .description("garbage collector roots directory")

  + .labels({"path"})

  + .dest(&gcRootsDir);
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 52
  [2.1006]

  [2.1006]
  + mkFlag()

  + .longName("dry-run")

  + .description("don't create store derivations")

  + .set(&dryRun, true);
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 57
  [2.1007]→[2.1007:1127](∅→∅)
  − static void findJobs(EvalState & state, JSONObject & top,

  − Bindings & autoArgs, Value & v, const string & attrPath);
  [2.1007]

  [2.1127]
  + mkFlag()

  + .longName("flake")

  + .description("build a flake")

  + .set(&flake, true);

  +

  + expectArg("expr", &releaseExpr);

  + }

  + };
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 66
  [2.1128]

  [2.1128]
  + static MyArgs myArgs;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 68
  [2.1129]→[2.1129:1244](∅→∅)
  − static string queryMetaStrings(EvalState & state, DrvInfo & drv, const string & name, const string & subAttribute)
  [2.1129]

  [2.1244]
  + static std::string queryMetaStrings(EvalState & state, DrvInfo & drv, const string & name, const string & subAttribute)
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 92
  [2.1884]→[2.1884:1885](∅→∅)
  −
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 93
  [2.1886]→[2.1886:2014](∅→∅)
  − static void findJobsWrapped(EvalState & state, JSONObject & top,

  − Bindings & autoArgs, Value & vIn, const string & attrPath)
  [2.1886]

  [2.2014]
  + static void worker(

  + EvalState & state,

  + Bindings & autoArgs,

  + AutoCloseFD & to,

  + AutoCloseFD & from)
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 99
  [2.2016]→[2.2016:2063](∅→∅)
  − debug(format("at path `%1%'") % attrPath);
  [2.2016]

  [2.2063]
  + Value vTop;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 101
  [2.2064]→[2.2064:2086](∅→∅)
  − checkInterrupt();
  [2.2064]

  [2.2086]
  + if (myArgs.flake) {

  + using namespace flake;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 104
  [2.2087]→[2.2087:2146](∅→∅)
  − Value v;

  − state.autoCallFunction(autoArgs, vIn, v);
  [2.2087]

  [2.2146]
  + auto flakeRef = parseFlakeRef(myArgs.releaseExpr);
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 106
  [2.2147]→[2.2147:2175](∅→∅)
  − if (v.type == tAttrs) {
  [2.2147]

  [2.2175]
  + auto vFlake = state.allocValue();
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 108
  [2.2176]→[2.2176:2227](∅→∅)
  − auto drv = getDerivation(state, v, false);
  [2.2176]

  [2.2227]
  + auto lockedFlake = lockFlake(state, flakeRef,

  + LockFlags {

  + .updateLockFile = false,

  + .useRegistries = false,

  + .allowMutable = false,

  + });
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 115
  [2.2228]→[2.2228:2273](∅→∅)
  − if (drv) {

  − Path drvPath;
  [2.2228]

  [2.2273]
  + callFlake(state, lockedFlake, *vFlake);
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 117
  [2.2274]→[2.2274:2334](∅→∅)
  − DrvInfo::Outputs outputs = drv->queryOutputs();
  [2.2274]

  [2.2334]
  + auto vOutputs = vFlake->attrs->get(state.symbols.create("outputs"))->value;

  + state.forceValue(*vOutputs);

  +

  + auto aHydraJobs = vOutputs->attrs->get(state.symbols.create("hydraJobs"));

  + if (!aHydraJobs)

  + aHydraJobs = vOutputs->attrs->get(state.symbols.create("checks"));

  + if (!aHydraJobs)

  + throw Error("flake '%s' does not provide any Hydra jobs or checks", flakeRef);

  +

  + vTop = *aHydraJobs->value;

  +

  + } else {

  + state.evalFile(lookupFileArg(state, myArgs.releaseExpr), vTop);

  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 132
  [2.2335]→[2.2335:2466](∅→∅)
  − if (drv->querySystem() == "unknown")

  − throw EvalError("derivation must have a ‘system’ attribute");
  [2.2335]

  [2.2466]
  + auto vRoot = state.allocValue();

  + state.autoCallFunction(autoArgs, vTop, *vRoot);
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 135
  [2.2467]→[2.2467:3328](∅→∅)
  − {

  − auto res = top.object(attrPath);

  − res.attr("nixName", drv->queryName());

  − res.attr("system", drv->querySystem());

  − res.attr("drvPath", drvPath = drv->queryDrvPath());

  − res.attr("description", drv->queryMetaString("description"));

  − res.attr("license", queryMetaStrings(state, *drv, "license", "shortName"));

  − res.attr("homepage", drv->queryMetaString("homepage"));

  − res.attr("maintainers", queryMetaStrings(state, *drv, "maintainers", "email"));

  − res.attr("schedulingPriority", drv->queryMetaInt("schedulingPriority", 100));

  − res.attr("timeout", drv->queryMetaInt("timeout", 36000));

  − res.attr("maxSilent", drv->queryMetaInt("maxSilent", 7200));

  − res.attr("isChannel", drv->queryMetaBool("isHydraChannel", false));
  [2.2467]

  [2.3328]
  + while (true) {

  + /* Wait for the master to send us a job name. */

  + writeLine(to.get(), "next");
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 139
  [2.3329]→[2.3329:4133](∅→∅)
  − /* If this is an aggregate, then get its constituents. */

  − Bindings::iterator a = v.attrs->find(state.symbols.create("_hydraAggregate"));

  − if (a != v.attrs->end() && state.forceBool(*a->value, *a->pos)) {

  − Bindings::iterator a = v.attrs->find(state.symbols.create("constituents"));

  − if (a == v.attrs->end())

  − throw EvalError("derivation must have a ‘constituents’ attribute");

  − PathSet context;

  − state.coerceToString(*a->pos, *a->value, context, true, false);

  − PathSet drvs;

  − for (auto & i : context)

  − if (i.at(0) == '!') {

  − size_t index = i.find("!", 1);

  − drvs.insert(string(i, index + 1));
  [2.3329]

  [2.4133]
  + auto s = readLine(from.get());

  + if (s == "exit") break;

  + if (!hasPrefix(s, "do ")) abort();

  + std::string attrPath(s, 3);

  +

  + debug("worker process %d at '%s'", getpid(), attrPath);

  +

  + /* Evaluate it and send info back to the master. */

  + nlohmann::json reply;

  +

  + try {

  + auto v = findAlongAttrPath(state, attrPath, autoArgs, *vRoot).first;

  +

  + state.forceValue(*v);

  +

  + if (auto drv = getDerivation(state, *v, false)) {

  +

  + DrvInfo::Outputs outputs = drv->queryOutputs();

  +

  + if (drv->querySystem() == "unknown")

  + throw EvalError("derivation must have a 'system' attribute");

  +

  + auto drvPath = drv->queryDrvPath();

  +

  + nlohmann::json job;

  +

  + job["nixName"] = drv->queryName();

  + job["system"] =drv->querySystem();

  + job["drvPath"] = drvPath;

  + job["description"] = drv->queryMetaString("description");

  + job["license"] = queryMetaStrings(state, *drv, "license", "shortName");

  + job["homepage"] = drv->queryMetaString("homepage");

  + job["maintainers"] = queryMetaStrings(state, *drv, "maintainers", "email");

  + job["schedulingPriority"] = drv->queryMetaInt("schedulingPriority", 100);

  + job["timeout"] = drv->queryMetaInt("timeout", 36000);

  + job["maxSilent"] = drv->queryMetaInt("maxSilent", 7200);

  + job["isChannel"] = drv->queryMetaBool("isHydraChannel", false);

  +

  + /* If this is an aggregate, then get its constituents. */

  + auto a = v->attrs->get(state.symbols.create("_hydraAggregate"));

  + if (a && state.forceBool(*a->value, *a->pos)) {

  + auto a = v->attrs->get(state.symbols.create("constituents"));

  + if (!a)

  + throw EvalError("derivation must have a ‘constituents’ attribute");

  +

  +

  + PathSet context;

  + state.coerceToString(*a->pos, *a->value, context, true, false);

  + for (auto & i : context)

  + if (i.at(0) == '!') {

  + size_t index = i.find("!", 1);

  + job["constituents"].push_back(string(i, index + 1));

  + }

  +

  + state.forceList(*a->value, *a->pos);

  + for (unsigned int n = 0; n < a->value->listSize(); ++n) {

  + auto v = a->value->listElems()[n];

  + state.forceValue(*v);

  + if (v->type == tString)

  + job["namedConstituents"].push_back(state.forceStringNoCtx(*v));
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 200
  [2.4155]→[2.4155:4240](∅→∅)
  − res.attr("constituents", concatStringsSep(" ", drvs));

  − }
  [2.4155]

  [2.4240]
  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 202
  [2.4241]→[2.4241:4752](∅→∅)
  − /* Register the derivation as a GC root. !!! This

  − registers roots for jobs that we may have already

  − done. */

  − auto localStore = state.store.dynamic_pointer_cast<LocalFSStore>();

  − if (gcRootsDir != "" && localStore) {

  − Path root = gcRootsDir + "/" + std::string(baseNameOf(drvPath));

  − if (!pathExists(root))

  − localStore->addPermRoot(localStore->parseStorePath(drvPath), root, false);

  − }
  [2.4241]

  [2.4752]
  + /* Register the derivation as a GC root. !!! This

  + registers roots for jobs that we may have already

  + done. */

  + auto localStore = state.store.dynamic_pointer_cast<LocalFSStore>();

  + if (gcRootsDir != "" && localStore) {

  + Path root = gcRootsDir + "/" + std::string(baseNameOf(drvPath));

  + if (!pathExists(root))

  + localStore->addPermRoot(localStore->parseStorePath(drvPath), root, false);

  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 212
  [2.4753]→[2.4753:4883](∅→∅)
  − auto res2 = res.object("outputs");

  − for (auto & j : outputs)

  − res2.attr(j.first, j.second);
  [2.4753]

  [2.4883]
  + nlohmann::json out;

  + for (auto & j : outputs)

  + out[j.first] = j.second;

  + job["outputs"] = std::move(out);
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 217
  [2.4884]

  [2.4884]
  + reply["job"] = std::move(job);
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 219
  [2.4898]→[2.4898:4908](∅→∅)
  − }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 220
  [2.4909]→[2.4909:5031](∅→∅)
  − else {

  − if (!state.isDerivation(v)) {

  − for (auto & i : v.attrs->lexicographicOrder()) {
  [2.4909]

  [2.5031]
  + else if (v->type == tAttrs) {

  + auto attrs = nlohmann::json::array();

  + StringSet ss;

  + for (auto & i : v->attrs->lexicographicOrder()) {
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 225
  [2.5078]→[2.5078:5201](∅→∅)
  −

  − /* Skip jobs with dots in the name. */

  − if (name.find('.') != std::string::npos) {
  [2.5078]

  [2.5201]
  + if (name.find('.') != std::string::npos || name.find(' ') != std::string::npos) {
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 229
  [2.5338]→[2.5338:5475](∅→∅)
  −

  − findJobs(state, top, autoArgs, *i->value,

  − (attrPath.empty() ? "" : attrPath + ".") + name);
  [2.5338]

  [2.5475]
  + attrs.push_back(name);
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 231
  [2.5493]

  [2.5493]
  + reply["attrs"] = std::move(attrs);
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 233
  [2.5507]

  [2.5507]
  +

  + } catch (EvalError & e) {

  + reply["error"] = filterANSIEscapes(e.msg(), true);
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 237
  [2.5517]→[2.5517:5523](∅→∅)
  − }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 238
  [2.5524]→[2.5524:5688](∅→∅)
  − else if (v.type == tNull) {

  − // allow null values, meaning 'do nothing'

  − }

  −

  − else

  − throw TypeError(format("unsupported value: %1%") % v);

  − }
  [2.5524]

  [2.5688]
  + writeLine(to.get(), reply.dump());
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 240
  [2.5689]→[2.5689:6013](∅→∅)
  −

  − static void findJobs(EvalState & state, JSONObject & top,

  − Bindings & autoArgs, Value & v, const string & attrPath)

  − {

  − try {

  − findJobsWrapped(state, top, autoArgs, v, attrPath);

  − } catch (EvalError & e) {

  − auto res = top.object(attrPath);

  − res.attr("error", filterANSIEscapes(e.msg(), true));
  [2.5689]

  [2.6013]
  + /* If our RSS exceeds the maximum, exit. The master will

  + start a new process. */

  + struct rusage r;

  + getrusage(RUSAGE_SELF, &r);

  + if ((size_t) r.ru_maxrss > maxMemorySize * 1024) break;
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 246
  [2.6019]→[2.6019:6021](∅→∅)
  − }
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 247
  [2.6022]

  [2.6022]
  + writeLine(to.get(), "restart");

  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 260
  [2.6267]→[2.6267:6462](∅→∅)
  − auto initialHeapSize = config->getStrOption("evaluator_initial_heap_size", "");

  − if (initialHeapSize != "")

  − setenv("GC_INITIAL_HEAP_SIZE", initialHeapSize.c_str(), 1);
  [2.6267]

  [2.6462]
  + auto nrWorkers = config->getIntOption("evaluator_workers", 1);

  + maxMemorySize = config->getIntOption("evaluator_max_memory_size", 4096);
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 265
  [2.6500]→[2.6500:6975](∅→∅)
  −

  − struct MyArgs : MixEvalArgs, MixCommonArgs

  − {

  − Path releaseExpr;

  − bool flake = false;

  −

  − MyArgs() : MixCommonArgs("hydra-eval-jobs")

  − {

  − mkFlag()

  − .longName("help")

  − .description("show usage information")

  − .handler([&]() {

  − printHelp(programName, std::cout);

  − throw Exit();

  − });
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 266
  [2.6976]→[2.6976:7639](∅→∅)
  − mkFlag()

  − .longName("gc-roots-dir")

  − .description("garbage collector roots directory")

  − .labels({"path"})

  − .dest(&gcRootsDir);

  −

  − mkFlag()

  − .longName("dry-run")

  − .description("don't create store derivations")

  − .set(&settings.readOnlyMode, true);

  −

  − mkFlag()

  − .longName("flake")

  − .description("build a flake")

  − .set(&flake, true);

  −

  − expectArg("expr", &releaseExpr);

  − }

  − };

  −

  − MyArgs myArgs;
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 267
  [2.7695]→[2.7695:7765](∅→∅)
  −

  − JSONObject json(std::cout, true);

  − std::cout.flush();
• edit in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 278
  [2.8224]

  [2.8224]
  +

  + if (myArgs.dryRun) settings.readOnlyMode = true;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 285
  [2.8403]→[2.8403:8460](∅→∅)
  − EvalState state(myArgs.searchPath, openStore());
  [2.8403]

  [2.8460]
  + struct State

  + {

  + std::set<std::string> todo{""};

  + std::set<std::string> active;

  + nlohmann::json jobs;

  + std::exception_ptr exc;

  + };
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 293
  [2.8461]→[2.8461:8519](∅→∅)
  − Bindings & autoArgs = *myArgs.getAutoArgs(state);
  [2.8461]

  [2.8519]
  + std::condition_variable wakeup;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 295
  [2.8520]→[2.8520:8537](∅→∅)
  − Value v;
  [2.8520]

  [2.8537]
  + Sync<State> state_;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 297
  [2.8538]→[2.8538:8601](∅→∅)
  − if (myArgs.flake) {

  − using namespace flake;
  [2.8538]

  [2.8601]
  + /* Start a handler thread per worker process. */

  + auto handler = [&]()

  + {

  + try {

  + pid_t pid = -1;

  + AutoCloseFD from, to;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 304
  [2.8602]→[2.8602:8665](∅→∅)
  − auto flakeRef = parseFlakeRef(myArgs.releaseExpr);
  [2.8602]

  [2.8665]
  + while (true) {
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 306
  [2.8666]→[2.8666:8712](∅→∅)
  − auto vFlake = state.allocValue();
  [2.8666]

  [2.8712]
  + /* Start a new worker process if necessary. */

  + if (pid == -1) {

  + Pipe toPipe, fromPipe;

  + toPipe.create();

  + fromPipe.create();

  + pid = startProcess(

  + [&,

  + to{std::make_shared<AutoCloseFD>(std::move(fromPipe.writeSide))},

  + from{std::make_shared<AutoCloseFD>(std::move(toPipe.readSide))}

  + ]()

  + {

  + try {

  + EvalState state(myArgs.searchPath, openStore());

  + Bindings & autoArgs = *myArgs.getAutoArgs(state);

  + worker(state, autoArgs, *to, *from);

  + } catch (std::exception & e) {

  + nlohmann::json err;

  + err["error"] = e.what();

  + writeLine(to->get(), err.dump());

  + }

  + },

  + ProcessOptions { .allowVfork = false });

  + from = std::move(fromPipe.readSide);

  + to = std::move(toPipe.writeSide);

  + debug("created worker process %d", pid);

  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 333
  [2.8713]→[2.8713:8951](∅→∅)
  − auto lockedFlake = lockFlake(state, flakeRef,

  − LockFlags {

  − .updateLockFile = false,

  − .useRegistries = false,

  − .allowMutable = false,

  − });
  [2.8713]

  [2.8951]
  + /* Check whether the existing worker process is still there. */

  + auto s = readLine(from.get());

  + if (s == "restart") {

  + pid = -1;

  + continue;

  + } else if (s != "next") {

  + auto json = nlohmann::json::parse(s);

  + throw Error("worker error: %s", (std::string) json["error"]);

  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 343
  [2.8952]→[2.8952:9004](∅→∅)
  − callFlake(state, lockedFlake, *vFlake);
  [2.8952]

  [2.9004]
  + /* Wait for a job name to become available. */

  + std::string attrPath;
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 346
  [2.9005]→[2.9005:9137](∅→∅)
  − auto vOutputs = (*vFlake->attrs->get(state.symbols.create("outputs")))->value;

  − state.forceValue(*vOutputs);
  [2.9005]

  [2.9137]
  + while (true) {

  + checkInterrupt();

  + auto state(state_.lock());

  + if ((state->todo.empty() && state->active.empty()) || state->exc) {

  + writeLine(to.get(), "exit");

  + return;

  + }

  + if (!state->todo.empty()) {

  + attrPath = *state->todo.begin();

  + state->todo.erase(state->todo.begin());

  + state->active.insert(attrPath);

  + break;

  + } else

  + state.wait(wakeup);

  + }

  +

  + /* Tell the worker to evaluate it. */

  + writeLine(to.get(), "do " + attrPath);

  +

  + /* Wait for the response. */

  + auto response = nlohmann::json::parse(readLine(from.get()));

  +

  + /* Handle the response. */

  + StringSet newAttrs;

  +

  + if (response.find("job") != response.end()) {

  + auto state(state_.lock());

  + state->jobs[attrPath] = response["job"];

  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 376
  [2.9138]→[2.9138:9461](∅→∅)
  − auto aHydraJobs = vOutputs->attrs->get(state.symbols.create("hydraJobs"));

  − if (!aHydraJobs)

  − aHydraJobs = vOutputs->attrs->get(state.symbols.create("checks"));

  − if (!aHydraJobs)

  − throw Error("flake '%s' does not provide any Hydra jobs or checks", flakeRef);
  [2.9138]

  [2.9461]
  + if (response.find("attrs") != response.end()) {

  + for (auto & i : response["attrs"]) {

  + auto s = (attrPath.empty() ? "" : attrPath + ".") + (std::string) i;

  + newAttrs.insert(s);

  + }

  + }

  +

  + if (response.find("error") != response.end()) {

  + auto state(state_.lock());

  + state->jobs[attrPath]["error"] = response["error"];

  + }
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 388
  [2.9462]→[2.9462:9501](∅→∅)
  − v = *(*aHydraJobs)->value;
  [2.9462]

  [2.9501]
  + /* Add newly discovered job names to the queue. */

  + {

  + auto state(state_.lock());

  + state->active.erase(attrPath);

  + for (auto & s : newAttrs)

  + state->todo.insert(s);

  + wakeup.notify_all();

  + }

  + }

  + } catch (...) {

  + auto state(state_.lock());

  + state->exc = std::current_exception();

  + wakeup.notify_all();

  + }

  + };
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 404
  [2.9502]→[2.9502:9592](∅→∅)
  − } else {

  − state.evalFile(lookupFileArg(state, myArgs.releaseExpr), v);
  [2.9502]

  [2.9592]
  + std::vector<std::thread> threads;

  + for (size_t i = 0; i < nrWorkers; i++)

  + threads.emplace_back(std::thread(handler));

  +

  + for (auto & thread : threads)

  + thread.join();

  +

  + auto state(state_.lock());

  +

  + if (state->exc)

  + std::rethrow_exception(state->exc);

  +

  + /* For aggregate jobs that have named consistuents

  + (i.e. constituents that are a job name rather than a

  + derivation), look up the referenced job and add it to the

  + dependencies of the aggregate derivation. */

  + auto store = openStore();

  +

  + for (auto i = state->jobs.begin(); i != state->jobs.end(); ++i) {

  + auto jobName = i.key();

  + auto & job = i.value();

  +

  + auto named = job.find("namedConstituents");

  + if (named == job.end()) continue;

  +

  + if (myArgs.dryRun) {

  + for (std::string jobName2 : *named) {

  + auto job2 = state->jobs.find(jobName2);

  + if (job2 == state->jobs.end())

  + throw Error("aggregate job '%s' references non-existent job '%s'", jobName, jobName2);

  + std::string drvPath2 = (*job2)["drvPath"];

  + job["constituents"].push_back(drvPath2);

  + }

  + } else {

  + std::string drvPath = job["drvPath"];

  + auto drv = readDerivation(*store, drvPath);

  +

  + for (std::string jobName2 : *named) {

  + auto job2 = state->jobs.find(jobName2);

  + if (job2 == state->jobs.end())

  + throw Error("aggregate job '%s' references non-existent job '%s'", jobName, jobName2);

  + std::string drvPath2 = (*job2)["drvPath"];

  + auto drv2 = readDerivation(*store, drvPath2);

  + job["constituents"].push_back(drvPath2);

  + drv.inputDrvs[store->parseStorePath(drvPath2)] = {drv2.outputs.begin()->first};

  + }

  +

  + std::string drvName(store->parseStorePath(drvPath).name());

  + assert(hasSuffix(drvName, drvExtension));

  + drvName.resize(drvName.size() - drvExtension.size());

  + auto h = hashDerivationModulo(*store, drv, true);

  + auto outPath = store->makeOutputPath("out", h, drvName);

  + drv.env["out"] = store->printStorePath(outPath);

  + drv.outputs.insert_or_assign("out", DerivationOutput(outPath.clone(), "", ""));

  + auto newDrvPath = store->printStorePath(writeDerivation(store, drv, drvName));

  +

  + debug("rewrote aggregate derivation %s -> %s", drvPath, newDrvPath);

  +

  + job["drvPath"] = newDrvPath;

  + job["outputs"]["out"] = store->printStorePath(outPath);

  + }

  +

  + job.erase("namedConstituents");
• replacement in src/hydra-eval-jobs/hydra-eval-jobs.cc at line 469
  [2.9603]→[2.9603:9651](∅→∅)
  − findJobs(state, json, autoArgs, v, "");
  [2.9603]

  [2.9651]
  + std::cout << state->jobs.dump(2) << "\n";