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

libutil: add Async{Context,IoRoot} to hide kj::AsyncIoContext

Browse Source 
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
This commit is contained in:
eldritch horrors
2025-01-24 13:48:44 +00:00
parent ae81f44b86
commit bd93b2750a
6 changed files with 81 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/build/derivation-goal.cc
+7 -7
View File
@@ -1404,7 +1404,7 @@ try {
auto buf = kj::heapArray<char>(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<char>(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<Outcome<void, Goal::WorkResult>> DerivationGoal::handleChildOutput()
try {
assert(builderOutFD);
auto builderIn = kj::heap<InputStream>(worker.aio.unixEventPort, builderOutFD->get());
auto builderIn = kj::heap<InputStream>(AIO().unixEventPort, builderOutFD->get());
kj::Own<InputStream> hookIn;
if (hook) {
hookIn = kj::heap<InputStream>(worker.aio.unixEventPort, hook->fromHook.get());
hookIn = kj::heap<InputStream>(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<void, WorkResult> {
return timedOut(
@@ -1523,7 +1523,7 @@ kj::Promise<Outcome<void, Goal::WorkResult>> 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<Outcome<void, Goal::WorkResult>> DerivationGoal::monitorForSilence()
kj::Promise<Outcome<void, Goal::WorkResult>>
DerivationGoal::handleChildStreams(InputStream & builderIn, InputStream * hookIn) noexcept
{
lastChildActivity = worker.aio.provider->getTimer().now();
lastChildActivity = AIO().provider.getTimer().now();
auto handlers = kj::joinPromisesFailFast([&] {
kj::Vector<kj::Promise<Outcome<void, WorkResult>>> parts{2};
lix/libstore/build/entry-points.cc
+22 -17
View File
@@ -8,14 +8,15 @@ namespace nix {
void Store::buildPaths(const std::vector<DerivedPath> & reqs, BuildMode buildMode, std::shared_ptr<Store> 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<Error> ex;
@@ -46,15 +47,16 @@ std::vector<KeyedBuildResult> Store::buildPathsWithResults(
BuildMode buildMode,
std::shared_ptr<Store> 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<KeyedBuildResult> results;
@@ -67,14 +69,15 @@ std::vector<KeyedBuildResult> 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));
}
lix/libstore/build/goal.cc
+1 -1
View File
@@ -17,7 +17,7 @@ kj::Promise<void> 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<Result<Goal::WorkResult>> Goal::work() noexcept
lix/libstore/build/worker.cc
+3 -4
View File
@@ -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<unsigned>(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<Result<Worker::Results>> 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 (...) {
lix/libstore/build/worker.hh
+7 -9
View File
@@ -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<uint64_t> 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<typename MkGoals>
friend kj::Promise<Result<Results>> processGoals(
Store & store, Store & evalStore, kj::AsyncIoContext & aio, MkGoals && mkGoals
) noexcept;
friend kj::Promise<Result<Results>>
processGoals(Store & store, Store & evalStore, MkGoals && mkGoals) noexcept;
};
template<typename MkGoals>
kj::Promise<Result<Worker::Results>> processGoals(
Store & store, Store & evalStore, kj::AsyncIoContext & aio, MkGoals && mkGoals
) noexcept
kj::Promise<Result<Worker::Results>>
processGoals(Store & store, Store & evalStore, MkGoals && mkGoals) noexcept
try {
co_return co_await Worker(store, evalStore, aio).run(std::forward<MkGoals>(mkGoals));
co_return co_await Worker(store, evalStore).run(std::forward<MkGoals>(mkGoals));
} catch (...) {
co_return result::current_exception();
}
lix/libutil/async.hh
+41
View File
@@ -2,8 +2,49 @@
///@file
#include "lix/libutil/result.hh"
#include <kj/async-io.h>
#include <kj/async.h>
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<void> r)
{