guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libstore, build-remote: delete static ssh:// fds

Browse Source 
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
This commit is contained in:
eldritch horrors
2025-07-09 18:21:20 +00:00
parent 11d854281a
commit 1a0d05d852
7 changed files with 97 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/legacy/build-remote.cc
+76 -14
View File
@@ -1,6 +1,10 @@
#include "lix/libutil/error.hh"
#include "lix/libutil/file-descriptor.hh"
#include "lix/libutil/logging.hh"
#include <algorithm>
#include <chrono>
#include <cstring>
#include <future>
#include <set>
#include <memory>
#include <string>
@@ -156,6 +160,61 @@ struct BuilderConnection
AutoCloseFD slotLock;
std::shared_ptr<Store> 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<void> startLogThread()
{
if (!logPipe.readSide) {
return {};
}
return std::async(
std::launch::async,
[](AutoCloseFD logFD) {
Activity act(*logger, lvlTalkative, actUnknown, "remote builder");
std::vector<char> 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<std::monostate, std::string, BuilderConnection> connectToBui
lock.reset();
std::shared_ptr<Store> 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<Store> sshStore;
AutoCloseFD bestSlotLock;
std::future<void> logThread;
std::optional<BuilderConnection> 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<StorePath> 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<std::string>(&result)) {
std::cerr << *immediateResponse;
continue;
} else {
builder = std::move(std::get<BuilderConnection>(result));
}
auto & builder = std::get<BuilderConnection>(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<PathSet>(source);
auto wantedOutputs = readStrings<StringSet>(source);
lix/libstore/build/derivation-goal.cc
+12 -10
View File
@@ -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<Outcome<void, Goal::WorkResult>> DerivationGoal::handleChildOutput() noexcept
try {
assert(builderOutFD);
auto builderIn = kj::heap<InputStream>(AIO().unixEventPort, builderOutFD->get());
kj::Own<InputStream> hookIn;
kj::Own<InputStream> builderIn, hookIn;
if (builderOutFD) {
builderIn = kj::heap<InputStream>(AIO().unixEventPort, builderOutFD->get());
}
if (hook) {
hookIn = kj::heap<InputStream>(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<Outcome<void, Goal::WorkResult>> DerivationGoal::monitorForSilence()
}
kj::Promise<Outcome<void, Goal::WorkResult>>
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<kj::Promise<Outcome<void, WorkResult>>> parts{2};
parts.add(handleBuilderOutput(builderIn));
if (builderIn) {
parts.add(handleBuilderOutput(*builderIn));
}
if (hookIn) {
parts.add(handleHookOutput(*hookIn));
}
lix/libstore/build/derivation-goal.hh
+1 -1
View File
@@ -317,7 +317,7 @@ protected:
kj::Promise<Outcome<void, WorkResult>> handleChildOutput() noexcept;
kj::Promise<Outcome<void, WorkResult>>
handleChildStreams(InputStream & builderIn, InputStream * hookIn) noexcept;
handleChildStreams(InputStream * builderIn, InputStream * hookIn) noexcept;
kj::Promise<Outcome<void, WorkResult>> handleBuilderOutput(InputStream & in) noexcept;
kj::Promise<Outcome<void, WorkResult>> handleHookOutput(InputStream & in) noexcept;
kj::Promise<Outcome<void, WorkResult>> monitorForSilence() noexcept;
lix/libstore/build/hook-instance.cc
-14
View File
@@ -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<FdSink>(toHook.get());
std::map<std::string, Config::SettingInfo> settings;
lix/libstore/build/hook-instance.hh
-5
View File
@@ -19,11 +19,6 @@ struct HookInstance
*/
AutoCloseFD fromHook;
/**
* Pipe for the builder's standard output/error.
*/
AutoCloseFD builderOut;
/**
* The process ID of the hook.
*/
lix/libstore/machines.cc
+6 -3
View File
@@ -65,11 +65,14 @@ bool Machine::mandatoryMet(const std::set<std::string> & features) const
});
}
kj::Promise<Result<ref<Store>>> Machine::openStore() const
kj::Promise<Result<std::pair<ref<Store>, 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();
}
lix/libstore/machines.hh
+2 -1
View File
@@ -1,6 +1,7 @@
#pragma once
///@file
#include "lix/libutil/file-descriptor.hh"
#include "lix/libutil/ref.hh"
#include "lix/libutil/result.hh"
#include <kj/async.h>
@@ -49,7 +50,7 @@ struct Machine {
decltype(mandatoryFeatures) mandatoryFeatures,
decltype(sshPublicHostKey) sshPublicHostKey);
kj::Promise<Result<ref<Store>>> openStore() const;
kj::Promise<Result<std::pair<ref<Store>, Pipe>>> openStore() const;
};
typedef std::vector<Machine> Machines;