diff --git a/lix/legacy/build-remote.cc b/lix/legacy/build-remote.cc index b33bbb5d3..48f8eed2b 100644 --- a/lix/legacy/build-remote.cc +++ b/lix/legacy/build-remote.cc @@ -1,4 +1,5 @@ #include "lix/libstore/path.hh" +#include "lix/libutil/async.hh" #include "lix/libutil/error.hh" #include "lix/libutil/file-descriptor.hh" #include "lix/libutil/logging.hh" @@ -183,7 +184,7 @@ struct BuilderConnection // 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() + std::future startLogThread(int intoFD) { if (!logPipe.readSide) { return {}; @@ -193,47 +194,16 @@ struct BuilderConnection 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; + [](int from, int to) { + AsyncIoRoot aio; - auto flushLine = [&] { - act.result(resBuildLogLine, {currentLogLine}); - currentLogLine.clear(); - currentLogLinePos = 0; - }; + auto reader = AIO().lowLevelProvider.wrapInputFd(from); + auto writer = AIO().lowLevelProvider.wrapOutputFd(to); - 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; - } - } - } + reader->pumpTo(*writer).wait(aio.kj.waitScope); }, - std::move(logPipe.readSide) + logPipe.readSide.get(), + intoFD ); } }; @@ -375,7 +345,7 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings auto conn = aio.kj.lowLevelProvider->wrapUnixSocketFd(1); capnp::TwoPartyServer srv(kj::heap(maxBuildJobs)); - srv.accept(*conn).wait(aio.kj.waitScope); + srv.accept(*conn, 1).wait(aio.kj.waitScope); return 0; } } @@ -442,7 +412,13 @@ kj::Promise Instance::build(BuildContext context) kj::Promise AcceptedBuild::run(RunContext context) { try { - auto logThread = builder.startLogThread(); + auto builderLogger = context.getParams().getBuildLogger(); + const int logFD = (co_await builderLogger.getFd()).orDefault(-1); + if (logFD < 0) { + throw Error("build-hook needs a logFD from the builder to build"); + } + + auto logThread = builder.startLogThread(logFD); KJ_DEFER({ // drop any existing ssh connection so the log thread can exit builder.sshStore = nullptr; diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index fc8640442..7eb93292d 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -1023,6 +1023,20 @@ try { co_return result::current_exception(); } +namespace { +struct BuildHookLogger final : rpc::build_remote::HookInstance::BuildLogger::Server +{ + AutoCloseFD fd; + + BuildHookLogger(AutoCloseFD fd) : fd(std::move(fd)) {} + + kj::Maybe getFd() override + { + return fd.get(); + } +}; +} + kj::Promise> DerivationGoal::tryBuildHook() try { if (!worker.hook.available || !useDerivation) { @@ -1039,6 +1053,12 @@ try { hook = TRY_AWAIT(HookInstance::create()); } + // open a pipe to receive logs directly from the hook + Pipe logPipe; + logPipe.create(); + builderOutFD = &logPipe.readSide; + KJ_DEFER(builderOutFD = nullptr); + KJ_DEFER(hook = nullptr); auto output = handleChildOutput(); @@ -1074,6 +1094,7 @@ try { /* Tell the hook all the inputs that have to be copied to the remote system. */ RPC_FILL(runReq, initInputs, inputPaths, worker.store); + runReq.setBuildLogger(kj::heap(std::move(logPipe.writeSide))); /* Tell the hooks the missing outputs that have to be copied back from the remote system. */ diff --git a/lix/libstore/build/hook-instance.capnp b/lix/libstore/build/hook-instance.capnp index b323edce3..81c8e962a 100644 --- a/lix/libstore/build/hook-instance.capnp +++ b/lix/libstore/build/hook-instance.capnp @@ -8,10 +8,15 @@ using Types = import "/lix/libutil/types.capnp"; using StoreTypes = import "/lix/libstore/types.capnp"; interface HookInstance { + interface BuildLogger { + # only used for fd passing + } + interface AcceptedBuild { run @0 ( inputs :List(StoreTypes.StorePath), # actual a set wantedOutputs :List(Data), # actually StringSet + buildLogger :BuildLogger, ) -> (result :Types.ResultV); } diff --git a/lix/libstore/build/hook-instance.hh b/lix/libstore/build/hook-instance.hh index e56f61f76..40f4990c1 100644 --- a/lix/libstore/build/hook-instance.hh +++ b/lix/libstore/build/hook-instance.hh @@ -6,6 +6,8 @@ #include "lix/libutil/processes.hh" #include "lix/libutil/serialise.hh" #include +#include +#include namespace nix { @@ -16,7 +18,7 @@ struct HookInstance */ AutoCloseFD fromHook; - kj::Own conn; + kj::Own conn; std::optional client; rpc::build_remote::HookInstance::Client rpc; @@ -32,7 +34,7 @@ struct HookInstance HookInstance(AutoCloseFD fromHook, AutoCloseFD rpc, Pid pid) : fromHook(std::move(fromHook)) , conn(AIO().lowLevelProvider.wrapUnixSocketFd(kj::AutoCloseFd(rpc.release()))) - , client(*this->conn) + , client(std::in_place, *this->conn, 1) , rpc(client->bootstrap().castAs()) , pid(std::move(pid)) {