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

libstore: convert build-hook protocol to rpc

Browse Source 
Change-Id: I8da74acdc965aba5091c089101745f7aa501befa
This commit is contained in:
eldritch horrors
2025-07-15 20:33:30 +02:00
parent 8a5a477ca3
commit 8f325fe436
10 changed files with 276 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/legacy/build-remote.cc
+122 -66
View File
@@ -1,7 +1,11 @@
#include "lix/libstore/path.hh"
#include "lix/libutil/error.hh"
#include "lix/libutil/file-descriptor.hh"
#include "lix/libutil/logging.hh"
#include "lix/libutil/rpc.hh"
#include "lix/libutil/types-rpc.hh"
#include <algorithm>
#include <capnp/rpc-twoparty.h>
#include <chrono>
#include <cstring>
#include <future>
@@ -24,13 +28,27 @@
#include "lix/libstore/derivations.hh"
#include "lix/libutil/strings.hh"
#include "lix/libstore/local-store.hh"
#include "lix/libstore/types-rpc.hh"
#include "lix/libcmd/legacy.hh"
#include "lix/libutil/experimental-features.hh"
#include "lix/libutil/hash.hh"
#include "build-remote.hh"
#include "lix/libstore/build/hook-instance.capnp.h"
namespace nix {
namespace {
struct Instance final : rpc::build_remote::HookInstance::Server
{
unsigned int maxBuildJobs;
Instance(unsigned int maxBuildJobs) : maxBuildJobs(maxBuildJobs) {}
kj::Promise<void> build(BuildContext context) override;
};
}
std::string escapeUri(std::string uri)
{
std::replace(uri.begin(), uri.end(), '/', '_');
@@ -171,6 +189,8 @@ struct BuilderConnection
return {};
}
logPipe.writeSide.close();
return std::async(
std::launch::async,
[](AutoCloseFD logFD) {
@@ -217,10 +237,27 @@ struct BuilderConnection
);
}
};
struct AcceptedBuild final : rpc::build_remote::HookInstance::AcceptedBuild::Server
{
ref<Store> store;
StorePath drvPath;
BuilderConnection builder;
AcceptedBuild(ref<Store> store, StorePath drvPath, BuilderConnection builder)
: store(store)
, drvPath(drvPath)
, builder(std::move(builder))
{
}
kj::Promise<void> run(RunContext context) override;
};
enum class BuildRejected { Temporarily, Permanently };
}
static kj::Promise<Result<std::variant<std::monostate, std::string, BuilderConnection>>>
connectToBuilder(
static kj::Promise<Result<std::variant<BuildRejected, BuilderConnection>>> connectToBuilder(
const ref<Store> & store,
const std::optional<StorePath> & drvPath,
Machines & machines,
@@ -255,7 +292,7 @@ try {
if (!bestSlotLock) {
if (rightType && !canBuildLocally) {
co_return "# postpone\n";
co_return BuildRejected::Temporarily;
} else {
printSelectionFailureMessage(
couldBuildLocally ? lvlChatty : lvlWarn,
@@ -265,7 +302,7 @@ try {
requiredFeatures
);
co_return "# decline\n";
co_return BuildRejected::Permanently;
}
}
@@ -301,8 +338,6 @@ try {
bestMachine->enabled = false;
}
}
co_return std::monostate{};
} catch (...) {
co_return result::current_exception();
}
@@ -338,13 +373,23 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
initPlugins();
auto conn = aio.kj.lowLevelProvider->wrapUnixSocketFd(1);
capnp::TwoPartyServer srv(kj::heap<Instance>(maxBuildJobs));
srv.accept(*conn).wait(aio.kj.waitScope);
return 0;
}
}
kj::Promise<void> Instance::build(BuildContext context)
{
try {
// FIXME this does not open a daemon connection for historical reasons.
// we may create a lot of build hook instances, and having each of them
// also create a daemon instance is inefficient and wasteful. in future
// versions of the build hook (where we don't need one hook process per
// build) we should change this to using a daemon connection, ideally a
// daemon connection provided by the parent via file descriptor passing
auto store = aio.blockOn(openStore(settings.storeUri, {}, AllowDaemon::Disallow));
auto store = TRY_AWAIT(openStore(settings.storeUri, {}, AllowDaemon::Disallow));
/* It would be more appropriate to use $XDG_RUNTIME_DIR, since
that gets cleared on reboot, but it wouldn't work on macOS. */
@@ -354,53 +399,63 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
else
currentLoad = settings.nixStateDir + currentLoadName;
std::future<void> logThread;
std::optional<BuilderConnection> builder;
auto machines = getMachines();
debug("got %d remote builders", machines.size());
if (machines.empty()) {
std::cerr << "# decline-permanently\n";
return 0;
context.getResults().initResult().initGood().setDeclinePermanently();
co_return;
}
std::optional<StorePath> drvPath;
auto amWilling = context.getParams().getAmWilling();
auto neededSystem = rpc::to<std::string>(context.getParams().getNeededSystem());
auto drvPath = from(context.getParams().getDrvPath(), *store);
auto requiredFeatures =
rpc::to<std::set<std::string>>(context.getParams().getRequiredFeatures());
while (!builder) {
auto result = TRY_AWAIT(connectToBuilder(
store, drvPath, machines, maxBuildJobs, amWilling, neededSystem, requiredFeatures
));
try {
auto s = readString(source);
if (s != "try") return 0;
} catch (EndOfFile &) { return 0; }
auto amWilling = readInt(source);
auto neededSystem = readString(source);
drvPath = store->parseStorePath(readString(source));
auto requiredFeatures = readStrings<std::set<std::string>>(source);
auto result = aio.blockOn(connectToBuilder(
store, drvPath, machines, maxBuildJobs, amWilling, neededSystem, requiredFeatures
));
if (std::get_if<std::monostate>(&result)) {
continue;
} else if (auto immediateResponse = std::get_if<std::string>(&result)) {
std::cerr << *immediateResponse;
} else {
builder = std::move(std::get<BuilderConnection>(result));
if (auto immediateResponse = std::get_if<BuildRejected>(&result)) {
switch (*immediateResponse) {
case BuildRejected::Temporarily:
context.getResults().initResult().initGood().setPostpone();
co_return;
case BuildRejected::Permanently:
context.getResults().initResult().initGood().setDecline();
co_return;
}
}
auto & sshStore = builder->sshStore;
auto & storeUri = builder->storeUri;
auto builder = std::get_if<BuilderConnection>(&result);
assert(builder);
std::cerr << "# accept\n" << storeUri << "\n";
auto ac = context.getResults().initResult().initGood().initAccept();
RPC_FILL(ac, setMachineName, builder->storeUri);
ac.setMachine(kj::heap<AcceptedBuild>(store, drvPath, std::move(*builder)));
} catch (...) {
RPC_FILL(context.getResults(), initResult, std::current_exception());
}
}
logThread = builder->startLogThread();
kj::Promise<void> AcceptedBuild::run(RunContext context)
{
try {
auto logThread = builder.startLogThread();
KJ_DEFER({
// drop any existing ssh connection so the log thread can exit
builder.sshStore = nullptr;
if (logThread.valid()) {
logThread.get();
}
});
auto inputs = readStrings<PathSet>(source);
auto wantedOutputs = readStrings<StringSet>(source);
auto & sshStore = builder.sshStore;
auto & storeUri = builder.storeUri;
auto inputs = rpc::to<std::set<StorePath>>(context.getParams().getInputs(), *store);
auto wantedOutputs = rpc::to<std::set<std::string>>(context.getParams().getWantedOutputs());
auto lockFileName = currentLoad + "/" + makeLockFilename(storeUri) + ".upload-lock";
@@ -409,7 +464,7 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
{
Activity act(*logger, lvlTalkative, actUnknown, fmt("waiting for the upload lock to '%s'", storeUri));
auto result = aio.blockOn(
auto result = TRY_AWAIT(
AIO().timeoutAfter(15 * kj::MINUTES, lockFileAsync(uploadLock.get(), ltWrite))
);
if (!result) {
@@ -421,19 +476,12 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
{
Activity act(*logger, lvlTalkative, actUnknown, fmt("copying dependencies to '%s'", storeUri));
aio.blockOn(copyPaths(
*store,
*sshStore,
store->parseStorePathSet(inputs),
NoRepair,
NoCheckSigs,
substitute
));
TRY_AWAIT(copyPaths(*store, *sshStore, inputs, NoRepair, NoCheckSigs, substitute));
}
uploadLock.reset();
auto drv = aio.blockOn(store->readDerivation(*drvPath));
auto drv = TRY_AWAIT(store->readDerivation(drvPath));
std::optional<BuildResult> optResult;
@@ -441,7 +489,7 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
// stores), we assume we are. This is necessary for backwards
// compat.
bool trustedOrLegacy = ({
std::optional trusted = aio.blockOn(sshStore->isTrustedClient());
std::optional trusted = TRY_AWAIT(sshStore->isTrustedClient());
!trusted || *trusted;
});
@@ -460,22 +508,27 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
//
// 2. Changing the `inputSrcs` set changes the associated
// output ids, which break CA derivations
if (!drv.inputDrvs.empty())
drv.inputSrcs = store->parseStorePathSet(inputs);
optResult = aio.blockOn(sshStore->buildDerivation(*drvPath, (const BasicDerivation &) drv));
if (!drv.inputDrvs.empty()) {
drv.inputSrcs = inputs;
}
optResult =
TRY_AWAIT(sshStore->buildDerivation(drvPath, (const BasicDerivation &) drv));
auto & result = *optResult;
if (!result.success())
throw Error("build of '%s' on '%s' failed: %s", store->printStorePath(*drvPath), storeUri, result.errorMsg);
throw Error(
"build of '%s' on '%s' failed: %s",
store->printStorePath(drvPath),
storeUri,
result.errorMsg
);
} else {
aio.blockOn(copyClosure(
*store, *sshStore, StorePathSet{*drvPath}, NoRepair, NoCheckSigs, substitute
TRY_AWAIT(copyClosure(
*store, *sshStore, StorePathSet{drvPath}, NoRepair, NoCheckSigs, substitute
));
auto res = aio.blockOn(sshStore->buildPathsWithResults({
DerivedPath::Built {
.drvPath = makeConstantStorePath(*drvPath),
.outputs = OutputsSpec::All {},
}
}));
auto res = TRY_AWAIT(sshStore->buildPathsWithResults({DerivedPath::Built{
.drvPath = makeConstantStorePath(drvPath),
.outputs = OutputsSpec::All{},
}}));
// One path to build should produce exactly one build result
assert(res.size() == 1);
optResult = std::move(res[0]);
@@ -485,8 +538,9 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
StorePathSet missingPaths;
auto outputPaths = drv.outputsAndPaths(*store);
for (auto & [outputName, outputPath] : outputPaths) {
if (!aio.blockOn(store->isValidPath(outputPath.second)))
if (!TRY_AWAIT(store->isValidPath(outputPath.second))) {
missingPaths.insert(outputPath.second);
}
}
if (!missingPaths.empty()) {
@@ -494,12 +548,14 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
if (auto localStore = store.try_cast_shared<LocalStore>())
for (auto & path : missingPaths)
localStore->locksHeld.insert(store->printStorePath(path)); /* FIXME: ugly */
aio.blockOn(
TRY_AWAIT(
copyPaths(*sshStore, *store, missingPaths, NoRepair, NoCheckSigs, NoSubstitute)
);
}
return 0;
context.getResults().initResult().setGood();
} catch (...) {
RPC_FILL(context.getResults(), initResult, std::current_exception());
}
}
lix/libstore/build/derivation-goal.cc
+51 -73
View File
@@ -12,9 +12,14 @@
#include "lix/libstore/local-store.hh" // TODO remove, along with remaining downcasts
#include "lix/libstore/build/substitution-goal.hh"
#include "lix/libutil/result.hh"
#include "lix/libutil/rpc.hh"
#include "lix/libutil/strings.hh"
#include "lix/libstore/build/hook-instance.capnp.h"
#include "lix/libstore/types-rpc.hh"
#include "lix/libutil/types-rpc.hh"
#include <boost/outcome/try.hpp>
#include <capnp/rpc-twoparty.h>
#include <fstream>
#include <kj/array.h>
#include <kj/async-unix.h>
@@ -22,6 +27,7 @@
#include <kj/debug.h>
#include <kj/vector.h>
#include <optional>
#include <ranges>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
@@ -1023,79 +1029,45 @@ try {
co_return HookResult::Decline{};
}
if (!worker.hook.instance)
worker.hook.instance = std::make_unique<HookInstance>();
try {
/* Send the request to the hook. */
*worker.hook.instance->sink << "try" << (slotToken.valid() ? 1 : 0) << drv->platform
<< worker.store.printStorePath(drvPath)
<< parsedDrv->getRequiredSystemFeatures();
worker.hook.instance->sink->flush();
/* Read the first line of input, which should be a word indicating
whether the hook wishes to perform the build. */
std::string reply;
while (true) {
auto s = [&]() {
try {
return readLine(worker.hook.instance->fromHook.get());
} catch (Error & e) {
e.addTrace({}, "while reading the response from the build hook");
throw;
}
}();
if (handleJSONLogMessage(s, worker.act, worker.hook.instance->activities, "the build hook", true))
;
else if (s.substr(0, 2) == "# ") {
reply = s.substr(2);
break;
}
else {
s += "\n";
writeLogsToStderr(s);
logger->log(lvlInfo, s);
}
}
debug("hook reply is '%1%'", reply);
if (reply == "decline")
co_return HookResult::Decline{};
else if (reply == "decline-permanently") {
worker.hook.available = false;
worker.hook.instance.reset();
co_return HookResult::Decline{};
}
else if (reply == "postpone")
co_return HookResult::Postpone{};
else if (reply != "accept")
throw Error("bad hook reply '%s'", reply);
} catch (SysError & e) {
if (e.errNo == EPIPE) {
printError(
"build hook died unexpectedly: %s",
chomp(drainFD(worker.hook.instance->fromHook.get())));
worker.hook.instance.reset();
co_return HookResult::Decline{};
} else
throw;
if (!worker.hook.instances.empty()) {
hook = std::move(worker.hook.instances.front());
worker.hook.instances.pop_front();
} else {
hook = TRY_AWAIT(HookInstance::create());
}
hook = std::move(worker.hook.instance);
KJ_DEFER(hook = nullptr);
auto output = handleChildOutput();
try {
machineName = readLine(hook->fromHook.get());
} catch (Error & e) {
e.addTrace({}, "while reading the machine name from the build hook");
throw;
auto buildReq = hook->rpc.buildRequest();
RPC_FILL(buildReq, setAmWilling, slotToken.valid());
RPC_FILL(buildReq, setNeededSystem, drv->platform);
RPC_FILL(buildReq, initDrvPath, drvPath, worker.store);
RPC_FILL(buildReq, initRequiredFeatures, parsedDrv->getRequiredSystemFeatures());
auto buildRespPromise = buildReq.send();
auto buildResp = TRY_AWAIT_RPC(buildRespPromise);
debug("hook reply is '%1%'", buildResp.toString().flatten().cStr());
if (buildResp.isDecline()) {
worker.hook.instances.push_back(std::move(hook));
co_return HookResult::Decline{};
} else if (buildResp.isDeclinePermanently()) {
worker.hook.available = false;
co_return HookResult::Decline{};
} else if (buildResp.isPostpone()) {
worker.hook.instances.push_back(std::move(hook));
co_return HookResult::Postpone{};
} else if (!buildResp.isAccept()) {
throw Error("bad hook reply '%s'", buildResp.which());
}
machineName = rpc::to<std::string>(buildResp.getAccept().getMachineName());
auto runReq = buildResp.getAccept().getMachine().runRequest();
/* Tell the hook all the inputs that have to be copied to the
remote system. */
*hook->sink << CommonProto::write({worker.store}, inputPaths);
RPC_FILL(runReq, initInputs, inputPaths, worker.store);
/* Tell the hooks the missing outputs that have to be copied back
from the remote system. */
@@ -1106,22 +1078,28 @@ try {
if (buildMode != bmCheck && status.known && status.known->isValid()) continue;
missingOutputs.insert(outputName);
}
*hook->sink << CommonProto::write({worker.store}, missingOutputs);
RPC_FILL(runReq, initWantedOutputs, missingOutputs);
}
hook->sink = nullptr;
hook->toHook.reset();
/* Create the log file and pipe. */
openLogFile();
// we have started to build. wait for the build to finish and process logs.
auto runPromise = runReq.send();
// build via hook is now properly running. wait for it to finish
actLock.reset();
buildResult.startTime = time(0); // inexact
started();
if (auto error = TRY_AWAIT(handleChildOutput())) {
co_return HookResult::Accept{*error};
TRY_AWAIT_RPC(runPromise);
// close the rpc connection to have the hook exit
hook->rpc = nullptr;
hook->client = std::nullopt;
hook->conn = nullptr;
if (auto error = TRY_AWAIT(output)) {
co_return HookResult::Accept{std::move(*error)};
}
co_return HookResult::Accept{TRY_AWAIT(buildDone())};
} catch (...) {
lix/libstore/build/derivation-goal.hh
+2 -1
View File
@@ -19,7 +19,8 @@ struct HookInstance;
struct HookResultBase
{
struct [[nodiscard]] Accept {
struct [[nodiscard]] Accept
{
Goal::WorkResult result;
};
struct [[nodiscard]] Decline {};
lix/libstore/build/hook-instance.capnp
+36
View File
@@ -0,0 +1,36 @@
@0xfa3817f907240eb2;
using Cxx = import "/capnp/c++.capnp";
$Cxx.namespace("nix::rpc::build_remote");
$Cxx.allowCancellation;
using Types = import "/lix/libutil/types.capnp";
using StoreTypes = import "/lix/libstore/types.capnp";
interface HookInstance {
interface AcceptedBuild {
run @0 (
inputs :List(StoreTypes.StorePath), # actual a set
wantedOutputs :List(Data), # actually StringSet
) -> (result :Types.ResultV);
}
struct BuildResponse {
union {
accept :group {
machine @0 :AcceptedBuild;
machineName @1 :Data;
}
postpone @2 :Void;
decline @3 :Void;
declinePermanently @4 :Void;
}
}
build @0 (
amWilling :Bool,
neededSystem :Data,
drvPath :StoreTypes.StorePath,
requiredFeatures :List(Data)
) -> (result :Types.Result(BuildResponse));
}
lix/libstore/build/hook-instance.cc
+22 -13
View File
@@ -7,8 +7,8 @@
namespace nix {
HookInstance::HookInstance()
{
kj::Promise<Result<std::unique_ptr<HookInstance>>> HookInstance::create()
try {
debug("starting build hook '%s'", concatStringsSep(" ", settings.buildHook.get()));
auto buildHookArgs = settings.buildHook.get();
@@ -35,9 +35,10 @@ HookInstance::HookInstance()
Pipe toHook_;
toHook_.create();
/* Fork the hook. */
pid = startProcess([&]() {
auto [selfRPC, hookRPC] = SocketPair::stream();
/* Fork the hook. */
auto pid = startProcess([&]() {
if (dup2(fromHook_.writeSide.get(), STDERR_FILENO) == -1)
throw SysError("cannot pipe standard error into log file");
@@ -46,8 +47,12 @@ HookInstance::HookInstance()
if (chdir("/") == -1) throw SysError("changing into /");
/* Dup the communication pipes. */
if (dup2(toHook_.readSide.get(), STDIN_FILENO) == -1)
if (dup2(toHook_.readSide.get(), STDIN_FILENO) == -1) {
throw SysError("dupping to-hook read side");
}
if (dup2(hookRPC.get(), STDOUT_FILENO) == -1) {
throw SysError("dupping to-hook read side");
}
execv(buildHook.c_str(), stringsToCharPtrs(args).data());
@@ -55,22 +60,26 @@ HookInstance::HookInstance()
});
pid.setSeparatePG(true);
fromHook = std::move(fromHook_.readSide);
toHook = std::move(toHook_.writeSide);
sink = std::make_unique<FdSink>(toHook.get());
std::map<std::string, Config::SettingInfo> settings;
globalConfig.getSettings(settings, true);
for (auto & setting : settings)
*sink << 1 << setting.first << setting.second.value;
*sink << 0;
}
FdSink sink(toHook_.writeSide.get());
for (auto & setting : settings) {
sink << 1 << setting.first << setting.second.value;
}
sink << 0;
sink.flush();
co_return std::make_unique<HookInstance>(
std::move(fromHook_.readSide), std::move(selfRPC), std::move(pid)
);
} catch (...) {
co_return result::current_exception();
}
HookInstance::~HookInstance()
{
try {
toHook.reset();
if (pid) pid.kill();
} catch (...) {
ignoreExceptionInDestructor();
lix/libstore/build/hook-instance.hh
+15 -9
View File
@@ -1,36 +1,42 @@
#pragma once
///@file
#include "hook-instance.capnp.h"
#include "lix/libutil/logging.hh"
#include "lix/libutil/processes.hh"
#include "lix/libutil/serialise.hh"
#include <capnp/rpc-twoparty.h>
namespace nix {
struct HookInstance
{
/**
* Pipe for talking to the build hook.
*/
AutoCloseFD toHook;
/**
* Pipe for the hook's standard output/error.
*/
AutoCloseFD fromHook;
kj::Own<kj::AsyncIoStream> conn;
std::optional<capnp::TwoPartyClient> client;
rpc::build_remote::HookInstance::Client rpc;
/**
* The process ID of the hook.
*/
Pid pid;
std::unique_ptr<FdSink> sink;
std::map<ActivityId, Activity> activities;
HookInstance();
static kj::Promise<Result<std::unique_ptr<HookInstance>>> create();
HookInstance(AutoCloseFD fromHook, AutoCloseFD rpc, Pid pid)
: fromHook(std::move(fromHook))
, conn(AIO().lowLevelProvider.wrapUnixSocketFd(kj::AutoCloseFd(rpc.release())))
, client(*this->conn)
, rpc(client->bootstrap().castAs<rpc::build_remote::HookInstance>())
, pid(std::move(pid))
{
}
~HookInstance();
};
}
lix/libstore/build/meson.build
+21
View File
@@ -0,0 +1,21 @@
libstore_rpc += custom_target(
command : [
capnpc_wrapper,
'--language=c++',
'--src-prefix=@CURRENT_SOURCE_DIR@',
'--outdir=@OUTDIR@',
'--depfile=@DEPFILE@',
'-I@SOURCE_ROOT@',
'@INPUT@',
],
input : files(
# keep-sorted start
'hook-instance.capnp',
# keep-sorted end
),
output : [
'@PLAINNAME@.h',
'@PLAINNAME@.c++',
],
depfile : '@PLAINNAME@.d',
)
lix/libstore/build/worker.hh
+3 -2
View File
@@ -202,8 +202,9 @@ private:
kj::TaskSet children;
public:
struct HookState {
std::unique_ptr<HookInstance> instance;
struct HookState
{
std::list<std::unique_ptr<HookInstance>> instances;
/**
* Whether to ask the build hook if it can build a derivation. If
lix/libstore/meson.build
+3
View File
@@ -35,6 +35,9 @@ libstore_rpc += custom_target(
depfile : '@PLAINNAME@.d',
)
# rpc definitions only here to get relative paths of generated files right
subdir('build')
foreach i : libstore_rpc
foreach rpc_src : i.to_list()
if rpc_src.full_path().endswith('.h')
lix/nix/meson.build
+1
View File
@@ -177,6 +177,7 @@ nix = executable(
boehm,
nlohmann_json,
kj,
capnp_rpc,
],
cpp_pch : cpp_pch,
install : true,