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
This commit is contained in:
Alois Wohlschlager
@@ -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<std::string>();
|
||||
}
|
||||
return joined;
|
||||
}
|
||||
|
||||
Proc(MyArgs &myArgs) {
|
||||
class Collector {
|
||||
struct Do {
|
||||
JSON attrPath;
|
||||
};
|
||||
struct Exit {};
|
||||
using Request = std::variant<Do, Exit>;
|
||||
|
||||
struct Next {};
|
||||
struct JsonResponse {
|
||||
JSON json;
|
||||
};
|
||||
struct Restart {};
|
||||
using Response = std::variant<Next, JsonResponse, Restart>;
|
||||
|
||||
RunningProgram child;
|
||||
AutoCloseFD to;
|
||||
std::optional<LineReader> 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<std::string, JSON> 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<std::string>();
|
||||
}
|
||||
return joined;
|
||||
}
|
||||
|
||||
void collector(MyArgs &myArgs, Sync<State> &state_,
|
||||
std::condition_variable &wakeup) {
|
||||
void collectorThread(MyArgs &myArgs, Sync<State> &state_, std::condition_variable &wakeup) {
|
||||
try {
|
||||
std::optional<std::unique_ptr<Proc>> proc_;
|
||||
std::optional<std::unique_ptr<LineReader>> fromReader_;
|
||||
Collector collector{myArgs.cmdline};
|
||||
|
||||
while (true) {
|
||||
if (!proc_.has_value()) {
|
||||
proc_ = std::make_unique<Proc>(myArgs);
|
||||
fromReader_ =
|
||||
std::make_unique<LineReader>(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> &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<JSON> newAttrs;
|
||||
@@ -260,9 +324,6 @@ void collector(MyArgs &myArgs, Sync<State> &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<std::thread> 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)));
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user