From 964ac8b0e88fb5789b87e33273e42363958d0afb Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Sat, 27 Apr 2024 01:02:22 +0200 Subject: libutil: de-callback-ify computeClosure only two users of this function exist. only one used it in a way that even bears resemblance to asynchronicity, and even that one didn't do it right. fully async and parallel computation would have only worked if any getEdgesAsync never calls the continuation it receives itself, only from more derived callbacks running on other threads. calling it directly would cause the decoupling promise to be awaited immediately *on the original thread*, completely negating all nice async effects. Change-Id: I0aa640950cf327533a32dee410105efdabb448df --- src/libstore/misc.cc | 32 +++++--------------- src/libstore/realisation.cc | 15 +--------- src/libutil/closure.hh | 68 +++++++++---------------------------------- tests/unit/libutil/closure.cc | 35 +++------------------- 4 files changed, 27 insertions(+), 123 deletions(-) diff --git a/src/libstore/misc.cc b/src/libstore/misc.cc index c8646083b..22f6b67ee 100644 --- a/src/libstore/misc.cc +++ b/src/libstore/misc.cc @@ -14,10 +14,9 @@ namespace nix { void Store::computeFSClosure(const StorePathSet & startPaths, StorePathSet & paths_, bool flipDirection, bool includeOutputs, bool includeDerivers) { - std::function(const StorePath & path, std::future> &)> queryDeps; + std::function(const StorePath & path, ref)> queryDeps; if (flipDirection) - queryDeps = [&](const StorePath& path, - std::future> & fut) { + queryDeps = [&](const StorePath& path, ref) { StorePathSet res; StorePathSet referrers; queryReferrers(path, referrers); @@ -36,10 +35,8 @@ void Store::computeFSClosure(const StorePathSet & startPaths, return res; }; else - queryDeps = [&](const StorePath& path, - std::future> & fut) { + queryDeps = [&](const StorePath& path, ref info) { StorePathSet res; - auto info = fut.get(); for (auto& ref : info->references) if (ref != path) res.insert(ref); @@ -54,24 +51,11 @@ void Store::computeFSClosure(const StorePathSet & startPaths, return res; }; - computeClosure( - startPaths, paths_, - [&](const StorePath& path, - std::function>&)> - processEdges) { - std::promise> promise; - std::function>)> - getDependencies = - [&](std::future> fut) { - try { - promise.set_value(queryDeps(path, fut)); - } catch (...) { - promise.set_exception(std::current_exception()); - } - }; - queryPathInfo(path, getDependencies); - processEdges(promise); - }); + paths_.merge(computeClosure( + startPaths, + [&](const StorePath& path) -> std::set { + return queryDeps(path, queryPathInfo(path)); + })); } void Store::computeFSClosure(const StorePath & startPath, diff --git a/src/libstore/realisation.cc b/src/libstore/realisation.cc index 93ddb5b20..50c911da0 100644 --- a/src/libstore/realisation.cc +++ b/src/libstore/realisation.cc @@ -43,20 +43,7 @@ void Realisation::closure(Store & store, const std::set & startOutp return res; }; - computeClosure( - startOutputs, res, - [&](const Realisation& current, - std::function>&)> - processEdges) { - std::promise> promise; - try { - auto res = getDeps(current); - promise.set_value(res); - } catch (...) { - promise.set_exception(std::current_exception()); - } - return processEdges(promise); - }); + res.merge(computeClosure(startOutputs, getDeps)); } nlohmann::json Realisation::toJSON() const { diff --git a/src/libutil/closure.hh b/src/libutil/closure.hh index 16e3b93e4..146931fc8 100644 --- a/src/libutil/closure.hh +++ b/src/libutil/closure.hh @@ -1,72 +1,32 @@ #pragma once ///@file +#include #include -#include -#include "sync.hh" - -using std::set; namespace nix { template -using GetEdgesAsync = std::function> &)>)>; - -template -void computeClosure( - const set startElts, - set & res, - GetEdgesAsync getEdgesAsync +std::set computeClosure( + std::set startElts, + std::function(const T &)> getEdges ) { - struct State - { - size_t pending; - set & res; - std::exception_ptr exc; - }; - - Sync state_(State{0, res, 0}); + std::set res, queue = std::move(startElts); - std::function enqueue; + while (!queue.empty()) { + std::set next; - std::condition_variable done; - - enqueue = [&](const T & current) -> void { - { - auto state(state_.lock()); - if (state->exc) return; - if (!state->res.insert(current).second) return; - state->pending++; + for (auto & e : queue) { + if (res.insert(e).second) { + next.merge(getEdges(e)); + } } - getEdgesAsync(current, [&](std::promise> & prom) { - try { - auto children = prom.get_future().get(); - for (auto & child : children) - enqueue(child); - { - auto state(state_.lock()); - assert(state->pending); - if (!--state->pending) done.notify_one(); - } - } catch (...) { - auto state(state_.lock()); - if (!state->exc) state->exc = std::current_exception(); - assert(state->pending); - if (!--state->pending) done.notify_one(); - }; - }); - }; - - for (auto & startElt : startElts) - enqueue(startElt); - - { - auto state(state_.lock()); - while (state->pending) state.wait(done); - if (state->exc) std::rethrow_exception(state->exc); + queue = std::move(next); } + + return res; } } diff --git a/tests/unit/libutil/closure.cc b/tests/unit/libutil/closure.cc index 7597e7807..b4eaad6f9 100644 --- a/tests/unit/libutil/closure.cc +++ b/tests/unit/libutil/closure.cc @@ -16,15 +16,11 @@ map> testGraph = { }; TEST(closure, correctClosure) { - set aClosure; set expectedClosure = {"A", "B", "C", "F", "G"}; - computeClosure( + set aClosure = computeClosure( {"A"}, - aClosure, - [&](const string currentNode, function> &)> processEdges) { - promise> promisedNodes; - promisedNodes.set_value(testGraph[currentNode]); - processEdges(promisedNodes); + [&](const string currentNode) { + return testGraph[currentNode]; } ); @@ -33,12 +29,10 @@ TEST(closure, correctClosure) { TEST(closure, properlyHandlesDirectExceptions) { struct TestExn {}; - set aClosure; EXPECT_THROW( computeClosure( {"A"}, - aClosure, - [&](const string currentNode, function> &)> processEdges) { + [&](const string currentNode) -> set { throw TestExn(); } ), @@ -46,25 +40,4 @@ TEST(closure, properlyHandlesDirectExceptions) { ); } -TEST(closure, properlyHandlesExceptionsInPromise) { - struct TestExn {}; - set aClosure; - EXPECT_THROW( - computeClosure( - {"A"}, - aClosure, - [&](const string currentNode, function> &)> processEdges) { - promise> promise; - try { - throw TestExn(); - } catch (...) { - promise.set_exception(std::current_exception()); - } - processEdges(promise); - } - ), - TestExn - ); -} - } -- cgit v1.2.3