From 1a0d05d85245a1e004b3049451fc20d96f85a891 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Wed, 9 Jul 2025 14:20:51 +0200 Subject: [PATCH] libstore, build-remote: delete static ssh:// fds this was a mess. ssh:// remotes used the extra static fds for build logs, ssh-ng:// remotes did not. ssh-ng remotes did not use them at all since ssh-ng never redirected them to begin with. we now create pipes dynamically and only for ssh:// builders, then translate logs received over these pipes into the same format used by ssh-ng. this requires a new activity we did not have before, but since we have a great many activities that rarely show up already this shouldn't be a problem for external tooling. if anything external tools can tell what's going on much better now (at least for ssh:// remote builds) Change-Id: I02010cee45598362a947faa3a5b04800d39daa31 --- lix/legacy/build-remote.cc | 90 ++++++++++++++++++++++----- lix/libstore/build/derivation-goal.cc | 22 ++++--- lix/libstore/build/derivation-goal.hh | 2 +- lix/libstore/build/hook-instance.cc | 14 ----- lix/libstore/build/hook-instance.hh | 5 -- lix/libstore/machines.cc | 9 ++- lix/libstore/machines.hh | 3 +- 7 files changed, 97 insertions(+), 48 deletions(-) diff --git a/lix/legacy/build-remote.cc b/lix/legacy/build-remote.cc index 3d0415aaf..b75d053f7 100644 --- a/lix/legacy/build-remote.cc +++ b/lix/legacy/build-remote.cc @@ -1,6 +1,10 @@ +#include "lix/libutil/error.hh" #include "lix/libutil/file-descriptor.hh" +#include "lix/libutil/logging.hh" #include #include +#include +#include #include #include #include @@ -156,6 +160,61 @@ struct BuilderConnection AutoCloseFD slotLock; std::shared_ptr sshStore; std::string storeUri; + Pipe logPipe; + + // start the thread that reads ssh stderr and turns it into log items. + // this future *must* outlive sshStore, otherwise it will never finish + std::future startLogThread() + { + if (!logPipe.readSide) { + return {}; + } + + return std::async( + std::launch::async, + [](AutoCloseFD logFD) { + Activity act(*logger, lvlTalkative, actUnknown, "remote builder"); + std::vector buf(4096); + size_t currentLogLinePos = 0; + std::string currentLogLine; + + auto flushLine = [&] { + act.result(resBuildLogLine, {currentLogLine}); + currentLogLine.clear(); + currentLogLinePos = 0; + }; + + while (true) { + const auto got = ::read(logFD.get(), buf.data(), buf.size()); + if (got < 0) { + printError("error reading builder response: %s", strerror(errno)); + break; + } else if (got == 0) { + if (!currentLogLine.empty()) { + flushLine(); + } + break; + } + + std::string_view data{buf.data(), size_t(got)}; + + for (auto c : data) { + if (c == '\r') { + currentLogLinePos = 0; + } else if (c == '\n') { + flushLine(); + } else { + if (currentLogLinePos >= currentLogLine.size()) { + currentLogLine.resize(currentLogLinePos + 1); + } + currentLogLine[currentLogLinePos++] = c; + } + } + } + }, + std::move(logPipe.readSide) + ); + } }; } @@ -217,16 +276,21 @@ static std::variant connectToBui lock.reset(); + std::shared_ptr sshStore; + Pipe logPipe; + try { Activity act( *logger, lvlTalkative, actUnknown, fmt("connecting to '%s'", bestMachine->storeUri) ); - auto sshStore = aio.blockOn(bestMachine->openStore()); + std::tie(sshStore, logPipe) = aio.blockOn(bestMachine->openStore()); aio.blockOn(sshStore->connect()); - return BuilderConnection{std::move(bestSlotLock), sshStore, bestMachine->storeUri}; + return BuilderConnection{ + std::move(bestSlotLock), sshStore, bestMachine->storeUri, std::move(logPipe) + }; } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) - auto msg = chomp(drainFD(5, false)); + std::string msg = logPipe.readSide ? chomp(drainFD(logPipe.readSide.get(), false)) : ""; printError( "cannot build on '%s': %s%s", bestMachine->storeUri, @@ -287,8 +351,8 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings else currentLoad = settings.nixStateDir + currentLoadName; - std::shared_ptr sshStore; - AutoCloseFD bestSlotLock; + std::future logThread; + std::optional builder; auto machines = getMachines(); debug("got %d remote builders", machines.size()); @@ -299,9 +363,8 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings } std::optional drvPath; - std::string storeUri; - while (!sshStore) { + while (!builder) { try { auto s = readString(source); @@ -328,19 +391,18 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings continue; } else if (auto immediateResponse = std::get_if(&result)) { std::cerr << *immediateResponse; - continue; + } else { + builder = std::move(std::get(result)); } - - auto & builder = std::get(result); - bestSlotLock = std::move(builder.slotLock); - sshStore = std::move(builder.sshStore); - storeUri = std::move(builder.storeUri); } - close(5); + auto & sshStore = builder->sshStore; + auto & storeUri = builder->storeUri; std::cerr << "# accept\n" << storeUri << "\n"; + logThread = builder->startLogThread(); + auto inputs = readStrings(source); auto wantedOutputs = readStrings(source); diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index 7f29e30bf..1aff6d4e2 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -801,12 +801,10 @@ int DerivationGoal::getChildStatus() void DerivationGoal::closeReadPipes() { - hook->builderOut.reset(); hook->fromHook.reset(); builderOutFD = nullptr; } - void DerivationGoal::cleanupHookFinally() { } @@ -1126,7 +1124,6 @@ HookReply DerivationGoal::tryBuildHook() /* Create the log file and pipe. */ openLogFile(); - builderOutFD = &hook->builderOut; return HookReply::Accept{handleChildOutput()}; } @@ -1319,15 +1316,16 @@ try { kj::Promise> DerivationGoal::handleChildOutput() noexcept try { - assert(builderOutFD); - - auto builderIn = kj::heap(AIO().unixEventPort, builderOutFD->get()); - kj::Own hookIn; + kj::Own builderIn, hookIn; + if (builderOutFD) { + builderIn = kj::heap(AIO().unixEventPort, builderOutFD->get()); + } if (hook) { hookIn = kj::heap(AIO().unixEventPort, hook->fromHook.get()); } - auto handlers = handleChildStreams(*builderIn, hookIn.get()).attach(std::move(builderIn), std::move(hookIn)); + auto handlers = handleChildStreams(builderIn.get(), hookIn.get()) + .attach(std::move(builderIn), std::move(hookIn)); if (respectsTimeouts() && settings.buildTimeout != 0) { handlers = handlers.exclusiveJoin( @@ -1364,14 +1362,18 @@ kj::Promise> DerivationGoal::monitorForSilence() } kj::Promise> -DerivationGoal::handleChildStreams(InputStream & builderIn, InputStream * hookIn) noexcept +DerivationGoal::handleChildStreams(InputStream * builderIn, InputStream * hookIn) noexcept { + assert(builderIn || hookIn); + lastChildActivity = AIO().provider.getTimer().now(); auto handlers = kj::joinPromisesFailFast([&] { kj::Vector>> parts{2}; - parts.add(handleBuilderOutput(builderIn)); + if (builderIn) { + parts.add(handleBuilderOutput(*builderIn)); + } if (hookIn) { parts.add(handleHookOutput(*hookIn)); } diff --git a/lix/libstore/build/derivation-goal.hh b/lix/libstore/build/derivation-goal.hh index 3fee2bf5b..fd1f64967 100644 --- a/lix/libstore/build/derivation-goal.hh +++ b/lix/libstore/build/derivation-goal.hh @@ -317,7 +317,7 @@ protected: kj::Promise> handleChildOutput() noexcept; kj::Promise> - handleChildStreams(InputStream & builderIn, InputStream * hookIn) noexcept; + handleChildStreams(InputStream * builderIn, InputStream * hookIn) noexcept; kj::Promise> handleBuilderOutput(InputStream & in) noexcept; kj::Promise> handleHookOutput(InputStream & in) noexcept; kj::Promise> monitorForSilence() noexcept; diff --git a/lix/libstore/build/hook-instance.cc b/lix/libstore/build/hook-instance.cc index 32adf893c..b50dfb67e 100644 --- a/lix/libstore/build/hook-instance.cc +++ b/lix/libstore/build/hook-instance.cc @@ -35,10 +35,6 @@ HookInstance::HookInstance() Pipe toHook_; toHook_.create(); - /* Create a pipe to get the output of the builder. */ - Pipe builderOut_; - builderOut_.create(); - /* Fork the hook. */ pid = startProcess([&]() { @@ -53,15 +49,6 @@ HookInstance::HookInstance() if (dup2(toHook_.readSide.get(), STDIN_FILENO) == -1) throw SysError("dupping to-hook read side"); - /* Use fd 4 for the builder's stdout/stderr. */ - if (dup2(builderOut_.writeSide.get(), 4) == -1) - throw SysError("dupping builder's stdout/stderr"); - - /* Hack: pass the read side of that fd to allow build-remote - to read SSH error messages. */ - if (dup2(builderOut_.readSide.get(), 5) == -1) - throw SysError("dupping builder's stdout/stderr"); - execv(buildHook.c_str(), stringsToCharPtrs(args).data()); throw SysError("executing '%s'", buildHook); @@ -70,7 +57,6 @@ HookInstance::HookInstance() pid.setSeparatePG(true); fromHook = std::move(fromHook_.readSide); toHook = std::move(toHook_.writeSide); - builderOut = std::move(builderOut_.readSide); sink = std::make_unique(toHook.get()); std::map settings; diff --git a/lix/libstore/build/hook-instance.hh b/lix/libstore/build/hook-instance.hh index 12841fd4f..52ff435d1 100644 --- a/lix/libstore/build/hook-instance.hh +++ b/lix/libstore/build/hook-instance.hh @@ -19,11 +19,6 @@ struct HookInstance */ AutoCloseFD fromHook; - /** - * Pipe for the builder's standard output/error. - */ - AutoCloseFD builderOut; - /** * The process ID of the hook. */ diff --git a/lix/libstore/machines.cc b/lix/libstore/machines.cc index b0caf60cc..daf91d532 100644 --- a/lix/libstore/machines.cc +++ b/lix/libstore/machines.cc @@ -65,11 +65,14 @@ bool Machine::mandatoryMet(const std::set & features) const }); } -kj::Promise>> Machine::openStore() const +kj::Promise, Pipe>>> Machine::openStore() const try { + Pipe pipe; + StoreConfig::Params storeParams; if (storeUri.starts_with("ssh://")) { - storeParams["log-fd"] = "4"; + pipe.create(); + storeParams["log-fd"] = std::to_string(pipe.writeSide.get()); storeParams["max-connections"] = "1"; } @@ -92,7 +95,7 @@ try { append(mandatoryFeatures); } - co_return TRY_AWAIT(nix::openStore(storeUri, storeParams)); + co_return {TRY_AWAIT(nix::openStore(storeUri, storeParams)), std::move(pipe)}; } catch (...) { co_return result::current_exception(); } diff --git a/lix/libstore/machines.hh b/lix/libstore/machines.hh index 162bcd8ec..2f986e4d3 100644 --- a/lix/libstore/machines.hh +++ b/lix/libstore/machines.hh @@ -1,6 +1,7 @@ #pragma once ///@file +#include "lix/libutil/file-descriptor.hh" #include "lix/libutil/ref.hh" #include "lix/libutil/result.hh" #include @@ -49,7 +50,7 @@ struct Machine { decltype(mandatoryFeatures) mandatoryFeatures, decltype(sshPublicHostKey) sshPublicHostKey); - kj::Promise>> openStore() const; + kj::Promise, Pipe>>> openStore() const; }; typedef std::vector Machines;