From bd93b2750ab5206bb43573fc90ac52a6cb7a37da Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Thu, 23 Jan 2025 23:38:07 +0100 Subject: [PATCH] libutil: add Async{Context,IoRoot} to hide kj::AsyncIoContext mostly we want to hide the wait scope since using a wait scope inside a promise is not allowed. we'll use this to push the AsyncIoContext up to the main() function as time passes, and presumably turn a lot of things that are regular synchronous code right now into promises as we proceed Change-Id: Ib972568a306b540b573e946492830ed5a15c485e --- lix/libstore/build/derivation-goal.cc | 14 ++++----- lix/libstore/build/entry-points.cc | 39 ++++++++++++++----------- lix/libstore/build/goal.cc | 2 +- lix/libstore/build/worker.cc | 7 ++--- lix/libstore/build/worker.hh | 16 +++++------ lix/libutil/async.hh | 41 +++++++++++++++++++++++++++ 6 files changed, 81 insertions(+), 38 deletions(-) diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index 1f93f4d26..9ac233010 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -1404,7 +1404,7 @@ try { auto buf = kj::heapArray(4096); while (true) { auto data = co_await in.read(buf); - lastChildActivity = worker.aio.provider->getTimer().now(); + lastChildActivity = AIO().provider.getTimer().now(); if (data.empty()) { co_return result::success(); @@ -1437,7 +1437,7 @@ try { auto buf = kj::heapArray(4096); while (true) { auto data = co_await in.read(buf); - lastChildActivity = worker.aio.provider->getTimer().now(); + lastChildActivity = AIO().provider.getTimer().now(); if (data.empty()) { co_return result::success(); @@ -1490,17 +1490,17 @@ kj::Promise> DerivationGoal::handleChildOutput() try { assert(builderOutFD); - auto builderIn = kj::heap(worker.aio.unixEventPort, builderOutFD->get()); + auto builderIn = kj::heap(AIO().unixEventPort, builderOutFD->get()); kj::Own hookIn; if (hook) { - hookIn = kj::heap(worker.aio.unixEventPort, hook->fromHook.get()); + hookIn = kj::heap(AIO().unixEventPort, hook->fromHook.get()); } auto handlers = handleChildStreams(*builderIn, hookIn.get()).attach(std::move(builderIn), std::move(hookIn)); if (respectsTimeouts() && settings.buildTimeout != 0) { handlers = handlers.exclusiveJoin( - worker.aio.provider->getTimer() + AIO().provider.getTimer() .afterDelay(settings.buildTimeout.get() * kj::SECONDS) .then([this]() -> Outcome { return timedOut( @@ -1523,7 +1523,7 @@ kj::Promise> DerivationGoal::monitorForSilence() while (true) { const auto stash = lastChildActivity; auto waitUntil = lastChildActivity + settings.maxSilentTime.get() * kj::SECONDS; - co_await worker.aio.provider->getTimer().atTime(waitUntil); + co_await AIO().provider.getTimer().atTime(waitUntil); if (lastChildActivity == stash) { co_return timedOut( Error("%1% timed out after %2% seconds of silence", name, settings.maxSilentTime) @@ -1535,7 +1535,7 @@ kj::Promise> DerivationGoal::monitorForSilence() kj::Promise> DerivationGoal::handleChildStreams(InputStream & builderIn, InputStream * hookIn) noexcept { - lastChildActivity = worker.aio.provider->getTimer().now(); + lastChildActivity = AIO().provider.getTimer().now(); auto handlers = kj::joinPromisesFailFast([&] { kj::Vector>> parts{2}; diff --git a/lix/libstore/build/entry-points.cc b/lix/libstore/build/entry-points.cc index 65e5fb29b..fc29aae39 100644 --- a/lix/libstore/build/entry-points.cc +++ b/lix/libstore/build/entry-points.cc @@ -8,14 +8,15 @@ namespace nix { void Store::buildPaths(const std::vector & reqs, BuildMode buildMode, std::shared_ptr evalStore) { - auto aio = kj::setupAsyncIo(); + auto kjaio = kj::setupAsyncIo(); + AsyncContext aio(kjaio); - auto results = processGoals(*this, evalStore ? *evalStore : *this, aio, [&](GoalFactory & gf) { + auto results = processGoals(*this, evalStore ? *evalStore : *this, [&](GoalFactory & gf) { Worker::Targets goals; for (auto & br : reqs) goals.emplace_back(gf.makeGoal(br, buildMode)); return goals; - }).wait(aio.waitScope).value(); + }).wait(kjaio.waitScope).value(); StringSet failed; std::shared_ptr ex; @@ -46,15 +47,16 @@ std::vector Store::buildPathsWithResults( BuildMode buildMode, std::shared_ptr evalStore) { - auto aio = kj::setupAsyncIo(); + auto kjaio = kj::setupAsyncIo(); + AsyncContext aio(kjaio); - auto goals = processGoals(*this, evalStore ? *evalStore : *this, aio, [&](GoalFactory & gf) { + auto goals = processGoals(*this, evalStore ? *evalStore : *this, [&](GoalFactory & gf) { Worker::Targets goals; for (const auto & req : reqs) { goals.emplace_back(gf.makeGoal(req, buildMode)); } return goals; - }).wait(aio.waitScope).value().goals; + }).wait(kjaio.waitScope).value().goals; std::vector results; @@ -67,14 +69,15 @@ std::vector Store::buildPathsWithResults( BuildResult Store::buildDerivation(const StorePath & drvPath, const BasicDerivation & drv, BuildMode buildMode) { - auto aio = kj::setupAsyncIo(); + auto kjaio = kj::setupAsyncIo(); + AsyncContext aio(kjaio); try { - auto results = processGoals(*this, *this, aio, [&](GoalFactory & gf) { + auto results = processGoals(*this, *this, [&](GoalFactory & gf) { Worker::Targets goals; goals.emplace_back(gf.makeBasicDerivationGoal(drvPath, drv, OutputsSpec::All{}, buildMode)); return goals; - }).wait(aio.waitScope).value(); + }).wait(kjaio.waitScope).value(); auto & result = results.goals.begin()->second; return result.result.restrictTo(DerivedPath::Built { .drvPath = makeConstantStorePathRef(drvPath), @@ -94,13 +97,14 @@ void Store::ensurePath(const StorePath & path) /* If the path is already valid, we're done. */ if (isValidPath(path)) return; - auto aio = kj::setupAsyncIo(); + auto kjaio = kj::setupAsyncIo(); + AsyncContext aio(kjaio); - auto results = processGoals(*this, *this, aio, [&](GoalFactory & gf) { + auto results = processGoals(*this, *this, [&](GoalFactory & gf) { Worker::Targets goals; goals.emplace_back(gf.makePathSubstitutionGoal(path)); return goals; - }).wait(aio.waitScope).value(); + }).wait(kjaio.waitScope).value(); auto & result = results.goals.begin()->second; if (result.exitCode != Goal::ecSuccess) { @@ -115,13 +119,14 @@ void Store::ensurePath(const StorePath & path) void Store::repairPath(const StorePath & path) { - auto aio = kj::setupAsyncIo(); + auto kjaio = kj::setupAsyncIo(); + AsyncContext aio(kjaio); - auto results = processGoals(*this, *this, aio, [&](GoalFactory & gf) { + auto results = processGoals(*this, *this, [&](GoalFactory & gf) { Worker::Targets goals; goals.emplace_back(gf.makePathSubstitutionGoal(path, Repair)); return goals; - }).wait(aio.waitScope).value(); + }).wait(kjaio.waitScope).value(); auto & result = results.goals.begin()->second; if (result.exitCode != Goal::ecSuccess) { @@ -129,7 +134,7 @@ void Store::repairPath(const StorePath & path) deriver, then rebuild the deriver. */ auto info = queryPathInfo(path); if (info->deriver && isValidPath(*info->deriver)) { - processGoals(*this, *this, aio, [&](GoalFactory & gf) { + processGoals(*this, *this, [&](GoalFactory & gf) { Worker::Targets goals; goals.emplace_back(gf.makeGoal( DerivedPath::Built{ @@ -140,7 +145,7 @@ void Store::repairPath(const StorePath & path) bmRepair )); return goals; - }).wait(aio.waitScope).value(); + }).wait(kjaio.waitScope).value(); } else throw Error(results.failingExitStatus, "cannot repair path '%s'", printStorePath(path)); } diff --git a/lix/libstore/build/goal.cc b/lix/libstore/build/goal.cc index 560ee0ba1..d5fdf4379 100644 --- a/lix/libstore/build/goal.cc +++ b/lix/libstore/build/goal.cc @@ -17,7 +17,7 @@ kj::Promise Goal::waitForAWhile() trace("wait for a while"); /* If we are polling goals that are waiting for a lock, then wake up after a few seconds at most. */ - return worker.aio.provider->getTimer().afterDelay(settings.pollInterval.get() * kj::SECONDS); + return AIO().provider.getTimer().afterDelay(settings.pollInterval.get() * kj::SECONDS); } kj::Promise> Goal::work() noexcept diff --git a/lix/libstore/build/worker.cc b/lix/libstore/build/worker.cc index c036e407d..ae23463b3 100644 --- a/lix/libstore/build/worker.cc +++ b/lix/libstore/build/worker.cc @@ -23,13 +23,12 @@ struct ErrorHandler : kj::TaskSet::ErrorHandler } errorHandler; } -Worker::Worker(Store & store, Store & evalStore, kj::AsyncIoContext & aio) +Worker::Worker(Store & store, Store & evalStore) : act(*logger, actRealise) , actDerivations(*logger, actBuilds) , actSubstitutions(*logger, actCopyPaths) , store(store) , evalStore(evalStore) - , aio(aio) /* Make sure that we are always allowed to run at least one substitution. This prevents infinite waiting. */ , substitutions(std::max(1, settings.maxSubstitutionJobs)) @@ -223,7 +222,7 @@ try { act.setExpected(actCopyPath, expectedNarSize + doneNarSize); // limit to 50fps. that should be more than good enough for anything we do - co_await aio.provider->getTimer().afterDelay(20 * kj::MILLISECONDS); + co_await AIO().provider.getTimer().afterDelay(20 * kj::MILLISECONDS); } } catch (...) { co_return result::current_exception(); @@ -314,7 +313,7 @@ try { kj::Promise> Worker::boopGC(LocalStore & localStore) try { while (true) { - co_await aio.provider->getTimer().afterDelay(10 * kj::SECONDS); + co_await AIO().provider.getTimer().afterDelay(10 * kj::SECONDS); localStore.autoGC(false); } } catch (...) { diff --git a/lix/libstore/build/worker.hh b/lix/libstore/build/worker.hh index 3e675dd97..c742c80ed 100644 --- a/lix/libstore/build/worker.hh +++ b/lix/libstore/build/worker.hh @@ -1,6 +1,7 @@ #pragma once ///@file +#include "lix/libutil/async.hh" #include "lix/libutil/async-semaphore.hh" #include "lix/libutil/concepts.hh" #include "lix/libutil/notifying-counter.hh" @@ -191,7 +192,6 @@ public: Store & store; Store & evalStore; - kj::AsyncIoContext & aio; AsyncSemaphore substitutions, localBuilds; private: @@ -225,7 +225,7 @@ public: NotifyingCounter doneNarSize{[this] { updateStatisticsLater(); }}; private: - Worker(Store & store, Store & evalStore, kj::AsyncIoContext & aio); + Worker(Store & store, Store & evalStore); ~Worker(); /** @@ -289,17 +289,15 @@ public: void markContentsGood(const StorePath & path); template - friend kj::Promise> processGoals( - Store & store, Store & evalStore, kj::AsyncIoContext & aio, MkGoals && mkGoals - ) noexcept; + friend kj::Promise> + processGoals(Store & store, Store & evalStore, MkGoals && mkGoals) noexcept; }; template -kj::Promise> processGoals( - Store & store, Store & evalStore, kj::AsyncIoContext & aio, MkGoals && mkGoals -) noexcept +kj::Promise> +processGoals(Store & store, Store & evalStore, MkGoals && mkGoals) noexcept try { - co_return co_await Worker(store, evalStore, aio).run(std::forward(mkGoals)); + co_return co_await Worker(store, evalStore).run(std::forward(mkGoals)); } catch (...) { co_return result::current_exception(); } diff --git a/lix/libutil/async.hh b/lix/libutil/async.hh index 72dd644fe..d96c1ee92 100644 --- a/lix/libutil/async.hh +++ b/lix/libutil/async.hh @@ -2,8 +2,49 @@ ///@file #include "lix/libutil/result.hh" +#include +#include namespace nix { + +struct AsyncContext +{ + static inline thread_local AsyncContext * current = nullptr; + + kj::AsyncIoProvider & provider; + kj::UnixEventPort & unixEventPort; + + explicit AsyncContext(kj::AsyncIoContext & aio) + : provider(*aio.provider) + , unixEventPort(aio.unixEventPort) + { + assert(current == nullptr); + current = this; + } + + ~AsyncContext() + { + current = nullptr; + } + + KJ_DISALLOW_COPY_AND_MOVE(AsyncContext); +}; + +struct AsyncIoRoot +{ + kj::AsyncIoContext kj; + AsyncContext context; + + AsyncIoRoot() : kj(kj::setupAsyncIo()), context(kj) {} + KJ_DISALLOW_COPY_AND_MOVE(AsyncIoRoot); +}; + +inline AsyncContext & AIO() +{ + assert(AsyncContext::current != nullptr); + return *AsyncContext::current; +} + namespace detail { inline void materializeResult(Result r) {