diff --git a/default.nix b/default.nix index 8fb058a46..9b398d61d 100644 --- a/default.nix +++ b/default.nix @@ -24,6 +24,8 @@ stdenv.mkDerivation { ninja # nlohmann_json can be only discovered via cmake files cmake + # XXX: ew + nix.passthru.capnproto-lix ] ++ (lib.optional stdenv.cc.isClang [ pkgs.clang-tools ]); meta = { diff --git a/flake.lock b/flake.lock index fa18ef32b..656b2cfc5 100644 --- a/flake.lock +++ b/flake.lock @@ -47,11 +47,11 @@ "pre-commit-hooks": "pre-commit-hooks" }, "locked": { - "lastModified": 1737234286, - "narHash": "sha256-pgDJZjj4jpzkFxsqBTI/9Yb0n3gW+DvDtuv9SwQZZcs=", - "rev": "079528098f5998ba13c88821a2eca1005c1695de", + "lastModified": 1737857294, + "narHash": "sha256-bzC+anLF/NlgolaMoB4uTFgSejLJlTzPcNF1Kbq/BP0=", + "rev": "4af6b5ed9f8f2412bef5331b8e3b93f3ad305ea1", "type": "tarball", - "url": "https://git.lix.systems/api/v1/repos/lix-project/lix/archive/079528098f5998ba13c88821a2eca1005c1695de.tar.gz?rev=079528098f5998ba13c88821a2eca1005c1695de" + "url": "https://git.lix.systems/api/v1/repos/lix-project/lix/archive/4af6b5ed9f8f2412bef5331b8e3b93f3ad305ea1.tar.gz?rev=4af6b5ed9f8f2412bef5331b8e3b93f3ad305ea1" }, "original": { "type": "tarball", @@ -96,11 +96,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1737226685, - "narHash": "sha256-34x0t/x5SkClo04gaG+KPBwN9JtXjFSQGTl//Yry4Gc=", + "lastModified": 1737672001, + "narHash": "sha256-YnHJJ19wqmibLQdUeq9xzE6CjrMA568KN/lFPuSVs4I=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "bf68d76e54ac5da0d77f82656a37552996195e80", + "rev": "035f8c0853c2977b24ffc4d0a42c74f00b182cd8", "type": "github" }, "original": { diff --git a/meson.build b/meson.build index 6cbdce64c..a4eede748 100644 --- a/meson.build +++ b/meson.build @@ -15,5 +15,6 @@ nix_cmd_dep = dependency('lix-cmd', required: true) threads_dep = dependency('threads', required: true) nlohmann_json_dep = dependency('nlohmann_json', required: true) boost_dep = dependency('boost', required: true) +kj_async_dep = dependency('kj-async', required: true) subdir('src') diff --git a/src/eval-args.cc b/src/eval-args.cc index b7c526afb..57b839c28 100644 --- a/src/eval-args.cc +++ b/src/eval-args.cc @@ -1,3 +1,4 @@ +#include #include #include #include @@ -11,7 +12,7 @@ #include "eval-args.hh" -MyArgs::MyArgs() : MixCommonArgs("nix-eval-jobs") { +MyArgs::MyArgs(nix::AsyncIoRoot & aio) : MixCommonArgs("nix-eval-jobs"), aio_(aio) { addFlag({ .longName = "help", .description = "show usage information", diff --git a/src/eval-args.hh b/src/eval-args.hh index b0932fb8c..1316c154f 100644 --- a/src/eval-args.hh +++ b/src/eval-args.hh @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -13,6 +14,10 @@ class MyArgs : virtual public nix::MixEvalArgs, virtual public nix::MixCommonArgs, virtual public nix::RootArgs { + // intentionally hidden in this subclass because it's mondo dangerous + // in n-e-j due to all the forking we do for worker process creation. + nix::AsyncIoRoot & aio_; + nix::AsyncIoRoot & aio() override { return aio_; } public: std::string releaseExpr; nix::Path gcRootsDir; @@ -31,7 +36,8 @@ class MyArgs : virtual public nix::MixEvalArgs, .writeLockFile = false, .useRegistries = false, .allowUnlocked = false}; - MyArgs(); + + MyArgs(nix::AsyncIoRoot & aio); MyArgs(const MyArgs&) = delete; void parseArgs(char** argv, int argc); diff --git a/src/nix-eval-jobs.cc b/src/nix-eval-jobs.cc index 8cba1c24d..8990398f5 100644 --- a/src/nix-eval-jobs.cc +++ b/src/nix-eval-jobs.cc @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -44,17 +45,16 @@ using namespace nix; using namespace nlohmann; -static MyArgs myArgs; - -using Processor = std::function state, Bindings &autoArgs, - AutoCloseFD &to, AutoCloseFD &from, MyArgs &args)>; +using Processor = std::function state, Bindings &autoArgs, + AutoCloseFD &to, AutoCloseFD &from, MyArgs &args, AsyncIoRoot &aio)>; /* Auto-cleanup of fork's process and fds. */ struct Proc { AutoCloseFD to, from; Pid pid; - Proc(const Processor &proc) { + Proc(MyArgs &myArgs, const Processor &proc) { Pipe toPipe, fromPipe; toPipe.create(); fromPipe.create(); @@ -65,13 +65,15 @@ struct Proc { std::make_shared(std::move(toPipe.readSide))}]() { debug("created worker process %d", getpid()); try { + AsyncIoRoot aio; auto evalStore = myArgs.evalStoreUrl ? openStore(*myArgs.evalStoreUrl) : openStore(); - auto evaluator = nix::make_ref(myArgs.searchPath, - evalStore); + auto evaluator = + nix::make_ref( + aio, myArgs.searchPath, evalStore); Bindings &autoArgs = *myArgs.getAutoArgs(*evaluator); - proc(evaluator, autoArgs, *to, *from, myArgs); + proc(evaluator, autoArgs, *to, *from, myArgs, aio); } catch (Error &e) { nlohmann::json err; auto msg = e.msg(); @@ -120,7 +122,8 @@ struct Thread { if ((s = pthread_attr_setstacksize(&attr, 64 * 1024 * 1024)) != 0) { throw SysError(s, "calling pthread_attr_setstacksize"); } - if ((s = pthread_create(&thread, &attr, Thread::init, func.release())) != 0) { + if ((s = pthread_create(&thread, &attr, Thread::init, + func.release())) != 0) { throw SysError(s, "calling pthread_launch"); } if ((s = pthread_attr_destroy(&attr)) != 0) { @@ -135,7 +138,8 @@ struct Thread { throw SysError(s, "calling pthread_join"); } } -private: + + private: static void *init(void *ptr) { std::unique_ptr> func; func.reset(static_cast *>(ptr)); @@ -221,14 +225,15 @@ std::string joinAttrPath(json &attrPath) { return joined; } -void collector(Sync &state_, std::condition_variable &wakeup) { +void collector(MyArgs &myArgs, Sync &state_, + std::condition_variable &wakeup) { try { std::optional> proc_; std::optional> fromReader_; while (true) { if (!proc_.has_value()) { - proc_ = std::make_unique(worker); + proc_ = std::make_unique(myArgs, worker); fromReader_ = std::make_unique(proc_.value()->from.release()); } @@ -344,6 +349,9 @@ int main(int argc, char **argv) { initNix(); initLibExpr(); + nix::AsyncIoRoot aio; + MyArgs myArgs(aio); + myArgs.parseArgs(argv, argc); /* When building a flake, use pure evaluation (no access to @@ -373,7 +381,8 @@ int main(int argc, char **argv) { std::vector threads; std::condition_variable wakeup; for (size_t i = 0; i < myArgs.nrWorkers; i++) { - threads.emplace_back(std::bind(collector, std::ref(state_), std::ref(wakeup))); + threads.emplace_back(std::bind(collector, std::ref(myArgs), + std::ref(state_), std::ref(wakeup))); } for (auto &thread : threads) diff --git a/src/worker.cc b/src/worker.cc index 9c8caa275..a11696809 100644 --- a/src/worker.cc +++ b/src/worker.cc @@ -49,8 +49,8 @@ static nix::Value *releaseExprTopLevelValue(nix::EvalState &state, nix::Value vTop; if (args.fromArgs) { - nix::Expr &e = state.ctx.parseExprFromString( - args.releaseExpr, nix::CanonPath::fromCwd()); + nix::Expr &e = state.ctx.parseExprFromString(args.releaseExpr, + nix::CanonPath::fromCwd()); state.eval(e, vTop); } else { state.evalFile(nix::lookupFileArg(state.ctx, args.releaseExpr), vTop); @@ -74,18 +74,19 @@ static std::string attrPathJoin(nlohmann::json input) { }); } -void worker(nix::ref evaluator, nix::Bindings &autoArgs, - nix::AutoCloseFD &to, nix::AutoCloseFD &from, MyArgs &args) { +void worker(nix::ref evaluator, + nix::Bindings &autoArgs, nix::AutoCloseFD &to, + nix::AutoCloseFD &from, MyArgs &args, nix::AsyncIoRoot &aio) { nix::Value *vRoot = [&]() { - auto state = evaluator->begin(); + auto state = evaluator->begin(aio); if (args.flake) { auto [flakeRef, fragment, outputSpec] = nix::parseFlakeRefWithFragmentAndExtendedOutputsSpec( args.releaseExpr, nix::absPath(".")); nix::InstallableFlake flake{ {}, evaluator, std::move(flakeRef), fragment, outputSpec, - {}, {}, args.lockFlags}; + {}, {}, args.lockFlags}; return flake.toValue(*state).first; } else { @@ -94,7 +95,7 @@ void worker(nix::ref evaluator, nix::Bindings }(); LineReader fromReader(from.release()); - auto state = evaluator->begin(); + auto state = evaluator->begin(aio); while (true) { /* Wait for the collector to send us a job name. */ @@ -160,8 +161,8 @@ void worker(nix::ref evaluator, nix::Bindings if (name == "recurseForDerivations" && !args.forceRecurse) { - auto attrv = - v->attrs->get(evaluator->s.recurseForDerivations); + auto attrv = v->attrs->get( + evaluator->s.recurseForDerivations); recurse = state->forceBool( *attrv->value, attrv->pos, "while evaluating recurseForDerivations"); diff --git a/src/worker.hh b/src/worker.hh index 013e64ada..d4c41caff 100644 --- a/src/worker.hh +++ b/src/worker.hh @@ -2,6 +2,7 @@ #include #include #include +#include #include "eval-args.hh" @@ -12,7 +13,8 @@ class AutoCloseFD; class Bindings; class EvalState; template class ref; -} // namespace nix +} // namespace nix -void worker(nix::ref evaluator, nix::Bindings &autoArgs, - nix::AutoCloseFD &to, nix::AutoCloseFD &from, MyArgs &args); +void worker(nix::ref evaluator, + nix::Bindings &autoArgs, nix::AutoCloseFD &to, + nix::AutoCloseFD &from, MyArgs &args, nix::AsyncIoRoot &aio);