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

fix for asyncized lix

Browse Source
This commit is contained in:
Jade Lovelace
2025-01-27 20:54:38 +01:00
committed by eldritch horrors
parent 6482bee40b
commit 53ce7ebc33
8 changed files with 56 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
default.nix
+2
View File
@@ -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 = {
flake.lock
Generated
+7 -7
View File
@@ -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": {
meson.build
+1
View File
@@ -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')
src/eval-args.cc
+2 -1
View File
@@ -1,3 +1,4 @@
#include <lix/libutil/async.hh>
#include <stdio.h>
#include <stdlib.h>
#include <lix/libutil/args.hh>
@@ -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",
src/eval-args.hh
+7 -1
View File
@@ -3,6 +3,7 @@
#include <lix/libexpr/flake/flake.hh>
#include <lix/libutil/args/root.hh>
#include <lix/libcmd/common-eval-args.hh>
#include <lix/libutil/async.hh>
#include <stddef.h>
#include <lix/libmain/common-args.hh>
#include <lix/libexpr/flake/flakeref.hh>
@@ -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);
src/nix-eval-jobs.cc
+22 -13
View File
@@ -2,6 +2,7 @@
#include <lix/libexpr/eval-settings.hh>
#include <lix/libmain/shared.hh>
#include <lix/libutil/async.hh>
#include <lix/libutil/sync.hh>
#include <lix/libexpr/eval.hh>
#include <lix/libutil/signals.hh>
@@ -44,17 +45,16 @@
using namespace nix;
using namespace nlohmann;
static MyArgs myArgs;
using Processor = std::function<void(ref<nix::eval_cache::CachingEvaluator> state, Bindings &autoArgs,
AutoCloseFD &to, AutoCloseFD &from, MyArgs &args)>;
using Processor = std::function<void(
ref<nix::eval_cache::CachingEvaluator> 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<AutoCloseFD>(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<nix::eval_cache::CachingEvaluator>(myArgs.searchPath,
evalStore);
auto evaluator =
nix::make_ref<nix::eval_cache::CachingEvaluator>(
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<std::function<void(void)>> func;
func.reset(static_cast<std::function<void(void)> *>(ptr));
@@ -221,14 +225,15 @@ std::string joinAttrPath(json &attrPath) {
return joined;
}
void collector(Sync<State> &state_, std::condition_variable &wakeup) {
void collector(MyArgs &myArgs, Sync<State> &state_,
std::condition_variable &wakeup) {
try {
std::optional<std::unique_ptr<Proc>> proc_;
std::optional<std::unique_ptr<LineReader>> fromReader_;
while (true) {
if (!proc_.has_value()) {
proc_ = std::make_unique<Proc>(worker);
proc_ = std::make_unique<Proc>(myArgs, worker);
fromReader_ =
std::make_unique<LineReader>(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<Thread> 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)
src/worker.cc
+10 -9
View File
@@ -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<nix::eval_cache::CachingEvaluator> evaluator, nix::Bindings &autoArgs,
nix::AutoCloseFD &to, nix::AutoCloseFD &from, MyArgs &args) {
void worker(nix::ref<nix::eval_cache::CachingEvaluator> 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<nix::eval_cache::CachingEvaluator> 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<nix::eval_cache::CachingEvaluator> 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");
src/worker.hh
+5 -3
View File
@@ -2,6 +2,7 @@
#include <lix/libmain/shared.hh>
#include <lix/libexpr/eval.hh>
#include <lix/libexpr/eval-cache.hh>
#include <lix/libutil/async.hh>
#include "eval-args.hh"
@@ -12,7 +13,8 @@ class AutoCloseFD;
class Bindings;
class EvalState;
template <typename T> class ref;
} // namespace nix
} // namespace nix
void worker(nix::ref<nix::eval_cache::CachingEvaluator> evaluator, nix::Bindings &autoArgs,
nix::AutoCloseFD &to, nix::AutoCloseFD &from, MyArgs &args);
void worker(nix::ref<nix::eval_cache::CachingEvaluator> evaluator,
nix::Bindings &autoArgs, nix::AutoCloseFD &to,
nix::AutoCloseFD &from, MyArgs &args, nix::AsyncIoRoot &aio);