From 7440c4ef751ef3017de140c197e98c5b2d68f324 Mon Sep 17 00:00:00 2001 From: Alois Wohlschlager Date: Mon, 13 Oct 2025 07:39:49 +0200 Subject: [PATCH] nix-eval-jobs: split the collector loop Previously the collector-side implementation of the worker interaction protocol was a bunch of spaghetti code. Split it apart at places where it makes sense to be easier to follow. Change-Id: If3cc6b6fbf289dd778856b40b55316c76a6a6964 --- .../nix-eval-jobs/src/nix-eval-jobs.cc | 331 +++++++++++------- 1 file changed, 196 insertions(+), 135 deletions(-) diff --git a/subprojects/nix-eval-jobs/src/nix-eval-jobs.cc b/subprojects/nix-eval-jobs/src/nix-eval-jobs.cc index f8a87fb86..8705e3d06 100644 --- a/subprojects/nix-eval-jobs/src/nix-eval-jobs.cc +++ b/subprojects/nix-eval-jobs/src/nix-eval-jobs.cc @@ -49,12 +49,38 @@ using namespace nix; static constexpr int NEJ_FROM_WORKER_FD = 3; static constexpr int NEJ_TO_WORKER_FD = 4; -/* Auto-cleanup of fork's process and fds. */ -struct Proc { - AutoCloseFD to, from; - RunningProgram child; +std::string joinAttrPath(const JSON &attrPath) { + std::string joined; + for (const auto &element : attrPath) { + if (!joined.empty()) { + joined += '.'; + } + joined += element.get(); + } + return joined; +} - Proc(MyArgs &myArgs) { +class Collector { + struct Do { + JSON attrPath; + }; + struct Exit {}; + using Request = std::variant; + + struct Next {}; + struct JsonResponse { + JSON json; + }; + struct Restart {}; + using Response = std::variant; + + RunningProgram child; + AutoCloseFD to; + std::optional from; + + Strings workerCmdline; + + void startWorker() { const auto self = [] { auto tmp = getSelfExe(); if (!tmp) { @@ -63,9 +89,6 @@ struct Proc { return *tmp; }(); - Strings args = myArgs.cmdline; - args.push_front("--worker"); - Pipe toPipe, fromPipe; toPipe.create(); fromPipe.create(); @@ -73,23 +96,176 @@ struct Proc { RunOptions options{ .program = self, .argv0 = "nix-eval-jobs", - .args = args, - .redirections = { - { .dup = NEJ_FROM_WORKER_FD, .from = fromPipe.writeSide.get() }, - { .dup = NEJ_TO_WORKER_FD, .from = toPipe.readSide.get() }, - }, + .args = workerCmdline, + .redirections = + { + {.dup = NEJ_FROM_WORKER_FD, .from = fromPipe.writeSide.get()}, + {.dup = NEJ_TO_WORKER_FD, .from = toPipe.readSide.get()}, + }, }; child = runProgram2(options); to = std::move(toPipe.writeSide); - from = std::move(fromPipe.readSide); + from.emplace(fromPipe.readSide.release()); } - ~Proc() { + void waitForWorkerReady() { + assert(child); + std::visit(overloaded{ + [](const Next &) {}, + [](const JsonResponse &response) { + throw Error("worker error: %s", (std::string) response.json["error"]); + }, + [&](const Restart &) { + (void) child.wait(); + to.reset(); + from.reset(); + }, + }, readResponse("checking worker process")); + } + + void makeWorkerReady() { + if (child) { + waitForWorkerReady(); + } + if (!child) { + startWorker(); + waitForWorkerReady(); + } + if (!child) { + throw Error("worker exited immediately"); + } + } + + Response readResponse(std::string_view msg) { + assert(from); + auto line = from->readLine(); + if (line.empty()) { + handleBrokenPipe(msg); + } else if (line == "next") { + return Next{}; + } else if (line == "restart") { + return Restart{}; + } else { + try { + return JsonResponse{JSON::parse(line)}; + } catch (const json::ParseError &e) { + throw Error( + "Received invalid JSON from worker: %s\n json: '%s'", + e.what(), line); + } + } + } + + void writeRequest(Request request) { + assert(to); + auto line = std::visit(overloaded{ + [](const Do &request) { + return fmt("do %s", request.attrPath.dump()); + }, + [](const Exit &) { + return std::string{"exit"}; + }, + }, request); + if (tryWriteLine(to.get(), line) < 0) { + auto msg = std::visit(overloaded{ + [](const Do &request) { + return fmt("sending attrPath '%s'", joinAttrPath(request.attrPath)); + }, + [](const Exit &) { + return std::string{"sending exit"}; + }, + }, request); + handleBrokenPipe(msg); + } + } + + [[noreturn]] void handleBrokenPipe(std::string_view msg) { + int status = child.wait(); + to.reset(); + from.reset(); + if (WIFEXITED(status)) { + if (WEXITSTATUS(status) == 0) { + // On the user hitting Ctrl-C, both the worker and the coordinator will receive the signal. + // When the worker is interrupted, it will "unexpectedly" exit successfully. + // Check whether the coordinator was interrupted as well, and don't show an ugly error in this case. + checkInterrupt(); + // Maybe the coordinator noticed the broken pipe before its own interrupt. + // Wait for a bit and try again. + std::this_thread::sleep_for(std::chrono::seconds(1)); + checkInterrupt(); + // No, the coordinator was not interrupted, possibly the signal was sent manually to the worker. + // Show the error in this case. + } else if (WEXITSTATUS(status) == 1) { + throw Error( + "while %s, evaluation worker exited with exit code 1, " + "(possible infinite recursion)", + msg); + } + throw Error("while %s, evaluation worker exited with %d", msg, + WEXITSTATUS(status)); + } else if (WIFSIGNALED(status)) { + switch (WTERMSIG(status)) { + case SIGKILL: + throw Error( + "while %s, evaluation worker got killed by SIGKILL, " + "maybe " + "memory limit reached?", + msg); + break; + case SIGSEGV: + throw Error( + "while %s, evaluation worker got killed by SIGSEGV, " + "(possible infinite recursion)", + msg); + default: + throw Error( + "while %s, evaluation worker got killed by signal %d (%s)", + msg, WTERMSIG(status), strsignal(WTERMSIG(status))); + } + } else { + // WIFSTOPPED and WIFCONTINUED should not happen, as neither WUNTRACED nor WCONTINUED are passed + throw Error("while %s, waitpid for evaluation worker returned unexpected status: %d", msg, status); + } + } + +public: + Collector(Strings cmdline) : workerCmdline{std::move(cmdline)} { + workerCmdline.push_front("--worker"); + } + + ~Collector() { if (child) { child.kill(); } } + + JSON evaluate(JSON attrPath) { + makeWorkerReady(); + writeRequest(Do{attrPath}); + return std::visit(overloaded{ + [](const Next &) -> JSON { + throw Error("unexpected response from worker: next"); + }, + [](const JsonResponse &response) { + return response.json; + }, + [](const Restart &) -> JSON { + throw Error("unexpected response from worker: restart"); + }, + }, readResponse(fmt("reading result for attrPath '%s'", joinAttrPath(attrPath)))); + } + + void exit() { + if (child) { + waitForWorkerReady(); + } + if (child) { + writeRequest(Exit{}); + // The worker will print "restart" when exiting cleanly, even if due to an explicit exit request. + waitForWorkerReady(); + } + } }; struct State { @@ -99,112 +275,19 @@ struct State { std::map jobs; }; -void handleBrokenWorkerPipe(Proc &proc, std::string_view msg) { - int status = proc.child.wait(); - if (WIFEXITED(status)) { - if (WEXITSTATUS(status) == 0) { - // On the user hitting Ctrl-C, both the worker and the coordinator will receive the signal. - // When the worker is interrupted, it will "unexpectedly" exit successfully. - // Check whether the coordinator was interrupted as well, and don't show an ugly error in this case. - checkInterrupt(); - // Maybe the coordinator noticed the broken pipe before its own interrupt. - // Wait for a bit and try again. - std::this_thread::sleep_for(std::chrono::seconds(1)); - checkInterrupt(); - // No, the coordinator was not interrupted, possibly the signal was sent manually to the worker. - // Show the error in this case. - } else if (WEXITSTATUS(status) == 1) { - throw Error( - "while %s, evaluation worker exited with exit code 1, " - "(possible infinite recursion)", - msg); - } - throw Error("while %s, evaluation worker exited with %d", msg, - WEXITSTATUS(status)); - } else if (WIFSIGNALED(status)) { - switch (WTERMSIG(status)) { - case SIGKILL: - throw Error( - "while %s, evaluation worker got killed by SIGKILL, " - "maybe " - "memory limit reached?", - msg); - break; - case SIGSEGV: - throw Error( - "while %s, evaluation worker got killed by SIGSEGV, " - "(possible infinite recursion)", - msg); - default: - throw Error( - "while %s, evaluation worker got killed by signal %d (%s)", - msg, WTERMSIG(status), strsignal(WTERMSIG(status))); - } - } else { - // WIFSTOPPED and WIFCONTINUED should not happen, as neither WUNTRACED nor WCONTINUED are passed - throw Error("while %s, waitpid for evaluation worker returned unexpected status: %d", msg, status); - } -} - -std::string joinAttrPath(JSON &attrPath) { - std::string joined; - for (auto &element : attrPath) { - if (!joined.empty()) { - joined += '.'; - } - joined += element.get(); - } - return joined; -} - -void collector(MyArgs &myArgs, Sync &state_, - std::condition_variable &wakeup) { +void collectorThread(MyArgs &myArgs, Sync &state_, std::condition_variable &wakeup) { try { - std::optional> proc_; - std::optional> fromReader_; + Collector collector{myArgs.cmdline}; while (true) { - if (!proc_.has_value()) { - proc_ = std::make_unique(myArgs); - fromReader_ = - std::make_unique(proc_.value()->from.release()); - } - auto proc = std::move(proc_.value()); - auto fromReader = std::move(fromReader_.value()); - - /* Check whether the existing worker process is still there. */ - auto s = fromReader->readLine(); - if (s.empty()) { - handleBrokenWorkerPipe(*proc.get(), "checking worker process"); - } else if (s == "restart") { - proc_ = std::nullopt; - fromReader_ = std::nullopt; - continue; - } else if (s != "next") { - try { - auto json = json::parse(s); - throw Error("worker error: %s", (std::string)json["error"]); - } catch (const json::ParseError &e) { - throw Error( - "Received invalid JSON from worker: %s\n json: '%s'", - e.what(), s); - } - } - /* Wait for a job name to become available. */ JSON attrPath; - while (true) { checkInterrupt(); auto state(state_.lock()); if ((state->todo.empty() && state->active.empty()) || state->exc) { - if (tryWriteLine(proc->to.get(), "exit") < 0) { - handleBrokenWorkerPipe(*proc.get(), "sending exit"); - } - if (proc->child) { - (void) proc->child.wait(); - } + collector.exit(); return; } if (!state->todo.empty()) { @@ -217,26 +300,7 @@ void collector(MyArgs &myArgs, Sync &state_, } /* Tell the worker to evaluate it. */ - if (tryWriteLine(proc->to.get(), "do " + attrPath.dump()) < 0) { - auto msg = "sending attrPath '" + joinAttrPath(attrPath) + "'"; - handleBrokenWorkerPipe(*proc.get(), msg); - } - - /* Wait for the response. */ - auto respString = fromReader->readLine(); - if (respString.empty()) { - auto msg = "reading result for attrPath '" + - joinAttrPath(attrPath) + "'"; - handleBrokenWorkerPipe(*proc.get(), msg); - } - JSON response; - try { - response = json::parse(respString); - } catch (const json::ParseError &e) { - throw Error( - "Received invalid JSON from worker: %s\n json: '%s'", - e.what(), respString); - } + auto response = collector.evaluate(attrPath); /* Handle the response. */ std::vector newAttrs; @@ -260,9 +324,6 @@ void collector(MyArgs &myArgs, Sync &state_, } } - proc_ = std::move(proc); - fromReader_ = std::move(fromReader); - /* Add newly discovered job names to the queue. */ { auto state(state_.lock()); @@ -335,7 +396,7 @@ int main(int argc, char **argv) { std::vector threads; std::condition_variable wakeup; for (size_t i = 0; i < myArgs.nrWorkers; i++) { - threads.emplace_back(std::bind(collector, std::ref(myArgs), + threads.emplace_back(std::bind(collectorThread, std::ref(myArgs), std::ref(state_), std::ref(wakeup))); }