From 4e266a25d6ec179705d79737768756a02ff74e99 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Wed, 5 Mar 2025 15:24:00 +0100 Subject: [PATCH] libutil/libstore: asyncify Pool connection pools of remote stores can currently block. this is not a problem when each request runs on a dedicated thread, but with async code this is no longer true. if a remote store has exhausted all its available connections on one executor and another job starts *on the same executor* we'll deadlock if that job makes another remote store request. unlike with sqlite previously it's not reasonable, not even necessary, to make the pools unbounded: since we use pools only with remote stores we can asyncify all of them at once, and since they're leaves of all call stacks we do not have much code to change either. Change-Id: I8c457e27893e22c2cfc35933307a5283c986805a --- lix/libstore/legacy-ssh-store.cc | 22 +++--- lix/libstore/local-store.cc | 28 ++++---- lix/libstore/remote-store.cc | 111 ++++++++++++++++--------------- lix/libstore/remote-store.hh | 2 +- lix/libstore/uds-remote-store.cc | 3 +- lix/libutil/pool.hh | 79 ++++++++++++++++------ tests/unit/libutil/pool.cc | 32 ++++++--- 7 files changed, 163 insertions(+), 114 deletions(-) diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc index 64bcea3b1..d5de5f19b 100644 --- a/lix/libstore/legacy-ssh-store.cc +++ b/lix/libstore/legacy-ssh-store.cc @@ -163,7 +163,7 @@ struct LegacySSHStore final : public Store kj::Promise>> queryPathInfoUncached(const StorePath & path) override try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); /* No longer support missing NAR hash */ assert(GET_PROTOCOL_MINOR(conn->remoteVersion) >= 4); @@ -197,7 +197,7 @@ struct LegacySSHStore final : public Store try { debug("adding path '%s' to remote host '%s'", printStorePath(info.path), host); - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 5) { @@ -253,7 +253,7 @@ struct LegacySSHStore final : public Store kj::Promise>> narFromPath(const StorePath & path) override try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); conn->to << ServeProto::Command::DumpStorePath << printStorePath(path); conn->to.flush(); @@ -319,7 +319,7 @@ public: const StorePath & drvPath, const BasicDerivation & drv, BuildMode buildMode ) override try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); conn->to << ServeProto::Command::BuildDerivation @@ -330,9 +330,9 @@ public: conn->to.flush(); - return {ServeProto::Serialise::read(*this, *conn)}; + co_return ServeProto::Serialise::read(*this, *conn); } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } kj ::Promise> buildPaths( @@ -344,7 +344,7 @@ public: if (evalStore && evalStore.get() != this) throw Error("building on an SSH store is incompatible with '--eval-store'"); - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); conn->to << ServeProto::Command::BuildPaths; Strings ss; @@ -409,7 +409,7 @@ public: co_return result::success(); } - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); conn->to << ServeProto::Command::QueryClosure @@ -427,7 +427,7 @@ public: kj::Promise> queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute = NoSubstitute) override try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); conn->to << ServeProto::Command::QueryValidPaths @@ -443,7 +443,7 @@ public: kj::Promise> connect() override try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -451,7 +451,7 @@ public: kj::Promise> getProtocol() override try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); co_return conn->remoteVersion; } catch (...) { co_return result::current_exception(); diff --git a/lix/libstore/local-store.cc b/lix/libstore/local-store.cc index fe2e7a056..ea2936139 100644 --- a/lix/libstore/local-store.cc +++ b/lix/libstore/local-store.cc @@ -792,7 +792,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) TRY_AWAIT(retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); if (auto oldR = queryRealisation_(*state, info.id)) { if (info.isCompatibleWith(*oldR)) { auto combinedSignatures = oldR->signatures; @@ -923,7 +923,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) retrySQLite([&]() -> kj::Promise>> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); co_return queryPathInfoInternal(*state, path); } catch (...) { co_return result::current_exception(); @@ -1019,7 +1019,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) co_return TRY_AWAIT(retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); co_return isValidPath_(*state, path); } catch (...) { co_return result::current_exception(); @@ -1047,7 +1047,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) co_return TRY_AWAIT(retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); auto use(state->stmts->QueryValidPaths.use()); StorePathSet res; while (use.next()) res.insert(parseStorePath(use.getStr(0))); @@ -1076,7 +1076,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) TRY_AWAIT(retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); queryReferrers(*state, path, referrers); co_return result::success(); } catch (...) { @@ -1094,7 +1094,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) co_return TRY_AWAIT(retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); auto useQueryValidDerivers(state->stmts->QueryValidDerivers.use()(printStorePath(path))); @@ -1119,7 +1119,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) retrySQLite([&]() -> kj::Promise>>> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); std::map> outputs; uint64_t drvId; drvId = queryValidPathId(*state, path); @@ -1147,7 +1147,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) co_return TRY_AWAIT(retrySQLite([&]() -> kj::Promise>> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); auto useQueryPathFromHashPart(state->stmts->QueryPathFromHashPart.use()(prefix)); @@ -1219,7 +1219,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) co_return co_await retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); SQLiteTxn txn = state->db.beginTransaction(SQLiteTxnType::Immediate); StorePathSet paths; @@ -1652,7 +1652,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) TRY_AWAIT(retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); SQLiteTxn txn = state->db.beginTransaction(SQLiteTxnType::Immediate); @@ -1783,7 +1783,7 @@ try { } if (update) { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); updatePathInfo(*state, *info); } @@ -1839,7 +1839,7 @@ try { if (canInvalidate) { printInfo("path '%s' disappeared, removing from database...", pathS); - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); TRY_AWAIT(invalidatePath(*state, path)); } else { printError("path '%s' disappeared, but it still has valid referrers!", pathS); @@ -1880,7 +1880,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) TRY_AWAIT(retrySQLite([&]() -> kj::Promise> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); SQLiteTxn txn = state->db.beginTransaction(SQLiteTxnType::Immediate); @@ -1988,7 +1988,7 @@ try { // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) TRY_AWAIT(retrySQLite([&]() -> kj::Promise>> { try { - auto state = dbPool.get(); + auto state = TRY_AWAIT(dbPool.get()); co_return queryRealisation_(*state, id); } catch (...) { co_return result::current_exception(); diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index c231f822d..d8a0995b6 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -190,14 +190,16 @@ void RemoteStore::ConnectionHandle::processStderr(Sink * sink, Source * source, } -RemoteStore::ConnectionHandle RemoteStore::getConnection() -{ - return ConnectionHandle(connections->get()); +kj::Promise> RemoteStore::getConnection() +try { + co_return ConnectionHandle(TRY_AWAIT(connections->get())); +} catch (...) { + co_return result::current_exception(); } kj::Promise> RemoteStore::setOptions() try { - setOptions(*(getConnection().handle)); + setOptions(*(TRY_AWAIT(getConnection()).handle)); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -205,7 +207,7 @@ try { kj::Promise> RemoteStore::isValidPathUncached(const StorePath & path) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::IsValidPath << printStorePath(path); conn.processStderr(); co_return readInt(conn->from); @@ -217,7 +219,7 @@ try { kj ::Promise> RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryValidPaths; conn->to << WorkerProto::write(*this, *conn, paths); if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 27) { @@ -232,7 +234,7 @@ try { kj::Promise> RemoteStore::queryAllValidPaths() try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryAllValidPaths; conn.processStderr(); co_return WorkerProto::Serialise::read(*this, *conn); @@ -243,21 +245,21 @@ try { kj::Promise> RemoteStore::querySubstitutablePaths(const StorePathSet & paths) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QuerySubstitutablePaths; conn->to << WorkerProto::write(*this, *conn, paths); conn.processStderr(); - return {WorkerProto::Serialise::read(*this, *conn)}; + co_return WorkerProto::Serialise::read(*this, *conn); } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } kj::Promise> RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, SubstitutablePathInfos & infos) try { - if (pathsMap.empty()) return {result::success()}; + if (pathsMap.empty()) co_return result::success(); - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QuerySubstitutablePathInfos; @@ -280,16 +282,16 @@ try { info.narSize = readLongLong(conn->from); } - return {result::success()}; + co_return result::success(); } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } kj::Promise>> RemoteStore::queryPathInfoUncached(const StorePath & path) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryPathInfo << printStorePath(path); try { conn.processStderr(); @@ -314,7 +316,7 @@ try { kj::Promise> RemoteStore::queryReferrers(const StorePath & path, StorePathSet & referrers) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryReferrers << printStorePath(path); conn.processStderr(); for (auto & i : WorkerProto::Serialise::read(*this, *conn)) @@ -327,7 +329,7 @@ try { kj::Promise> RemoteStore::queryValidDerivers(const StorePath & path) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryValidDerivers << printStorePath(path); conn.processStderr(); co_return WorkerProto::Serialise::read(*this, *conn); @@ -342,7 +344,7 @@ try { co_return TRY_AWAIT(Store::queryDerivationOutputs(path)); } REMOVE_AFTER_DROPPING_PROTO_MINOR(21); - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryDerivationOutputs << printStorePath(path); conn.processStderr(); co_return WorkerProto::Serialise::read(*this, *conn); @@ -356,7 +358,7 @@ RemoteStore::queryPartialDerivationOutputMap(const StorePath & path, Store * eva try { if (GET_PROTOCOL_MINOR(TRY_AWAIT(getProtocol())) >= 22) { if (!evalStore_) { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryDerivationOutputMap << printStorePath(path); conn.processStderr(); co_return WorkerProto::Serialise>>::read( @@ -395,7 +397,7 @@ try { kj::Promise>> RemoteStore::queryPathFromHashPart(const std::string & hashPart) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::QueryPathFromHashPart << hashPart; conn.processStderr(); Path path = readString(conn->from); @@ -414,7 +416,7 @@ kj::Promise>> RemoteStore::addCAToStore( const StorePathSet & references, RepairFlag repair) try { - std::optional conn_(getConnection()); + std::optional conn_(TRY_AWAIT(getConnection())); auto & conn = *conn_; if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 25) { @@ -529,7 +531,7 @@ kj::Promise> RemoteStore::addToStore( CheckSigsFlag checkSigs ) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::AddToStoreNar << printStorePath(info.path) @@ -569,10 +571,10 @@ kj::Promise> RemoteStore::addMultipleToStore( RepairFlag repair, CheckSigsFlag checkSigs) try { - if (GET_PROTOCOL_MINOR(getConnection()->daemonVersion) >= 32) { + if (GET_PROTOCOL_MINOR(TRY_AWAIT(getConnection())->daemonVersion) >= 32) { auto remoteVersion = TRY_AWAIT(getProtocol()); - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::AddMultipleToStore << repair @@ -618,7 +620,7 @@ try { kj::Promise> RemoteStore::registerDrvOutput(const Realisation & info) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::RegisterDrvOutput; if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) { REMOVE_AFTER_DROPPING_PROTO_MINOR(30); @@ -636,7 +638,7 @@ try { kj::Promise>> RemoteStore::queryRealisationUncached(const DrvOutput & id) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 27) { warn("the daemon is too old to support content-addressed derivations, please upgrade it to 2.4"); @@ -697,7 +699,7 @@ kj ::Promise> RemoteStore::buildPaths( try { TRY_AWAIT(copyDrvsFromEvalStore(drvPaths, evalStore)); - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::BuildPaths; conn->to << WorkerProto::write(*this, *conn, drvPaths); conn->to << buildMode; @@ -715,7 +717,7 @@ kj::Promise>> RemoteStore::buildPathsWithRe try { TRY_AWAIT(copyDrvsFromEvalStore(paths, evalStore)); - std::optional conn_(getConnection()); + std::optional conn_(TRY_AWAIT(getConnection())); auto & conn = *conn_; if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) { @@ -809,44 +811,44 @@ kj ::Promise> RemoteStore::buildDerivation( const StorePath & drvPath, const BasicDerivation & drv, BuildMode buildMode ) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::BuildDerivation << printStorePath(drvPath); writeDerivation(conn->to, *this, drv); conn->to << buildMode; conn.processStderr(); - return {WorkerProto::Serialise::read(*this, *conn)}; + co_return WorkerProto::Serialise::read(*this, *conn); } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } kj::Promise> RemoteStore::ensurePath(const StorePath & path) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::EnsurePath << printStorePath(path); conn.processStderr(); readInt(conn->from); - return {result::success()}; + co_return result::success(); } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } kj::Promise> RemoteStore::addTempRoot(const StorePath & path) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::AddTempRoot << printStorePath(path); conn.processStderr(); readInt(conn->from); - return {result::success()}; + co_return result::success(); } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } kj::Promise> RemoteStore::findRoots(bool censor) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::FindRoots; conn.processStderr(); size_t count = readNum(conn->from); @@ -865,7 +867,7 @@ try { kj::Promise> RemoteStore::collectGarbage(const GCOptions & options, GCResults & results) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::CollectGarbage << options.action; @@ -893,7 +895,7 @@ try { kj::Promise> RemoteStore::optimiseStore() try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::OptimiseStore; conn.processStderr(); readInt(conn->from); @@ -905,19 +907,19 @@ try { kj::Promise> RemoteStore::verifyStore(bool checkContents, RepairFlag repair) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::VerifyStore << checkContents << repair; conn.processStderr(); - return {readInt(conn->from)}; + co_return readInt(conn->from); } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } kj::Promise> RemoteStore::addSignatures(const StorePath & storePath, const StringSet & sigs) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::AddSignatures << printStorePath(storePath) << sigs; conn.processStderr(); readInt(conn->from); @@ -931,7 +933,7 @@ kj::Promise> RemoteStore::queryMissing(const std::vectorto << WorkerProto::Op::QueryMissing; conn->to << WorkerProto::write(*this, *conn, targets); conn.processStderr(); @@ -947,7 +949,7 @@ try { kj::Promise> RemoteStore::addBuildLog(const StorePath & drvPath, std::string_view log) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::AddBuildLog << drvPath.to_string(); StringSource source(log); conn.withFramedSink([&](Sink & sink) { @@ -962,7 +964,7 @@ try { kj::Promise>> RemoteStore::getVersion() try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); co_return conn->daemonNixVersion; } catch (...) { co_return result::current_exception(); @@ -971,7 +973,7 @@ try { kj::Promise> RemoteStore::connect() try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -980,7 +982,7 @@ try { kj::Promise> RemoteStore::getProtocol() try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); co_return conn->daemonVersion; } catch (...) { co_return result::current_exception(); @@ -988,11 +990,11 @@ try { kj::Promise>> RemoteStore::isTrustedClient() try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); co_return conn->remoteTrustsUs; } catch (...) { - co_return result::current_exception();} - + co_return result::current_exception(); +} RemoteStore::Connection::~Connection() @@ -1006,14 +1008,15 @@ RemoteStore::Connection::~Connection() kj::Promise>> RemoteStore::narFromPath(const StorePath & path) try { - auto conn(connections->get()); + auto conn(TRY_AWAIT(connections->get())); conn->to << WorkerProto::Op::NarFromPath << printStorePath(path); conn->processStderr(); co_return make_box_ptr([](auto conn) -> WireFormatGenerator { co_yield copyNAR(conn->from); }(std::move(conn))); } catch (...) { - co_return result::current_exception();} + co_return result::current_exception(); +} ref RemoteStore::getFSAccessor() diff --git a/lix/libstore/remote-store.hh b/lix/libstore/remote-store.hh index c07eeef25..56919fe36 100644 --- a/lix/libstore/remote-store.hh +++ b/lix/libstore/remote-store.hh @@ -194,7 +194,7 @@ protected: struct ConnectionHandle; - ConnectionHandle getConnection(); + kj::Promise> getConnection(); friend struct ConnectionHandle; diff --git a/lix/libstore/uds-remote-store.cc b/lix/libstore/uds-remote-store.cc index ff8f7c464..ca837f524 100644 --- a/lix/libstore/uds-remote-store.cc +++ b/lix/libstore/uds-remote-store.cc @@ -1,4 +1,5 @@ #include "lix/libstore/uds-remote-store.hh" +#include "lix/libutil/result.hh" #include "lix/libutil/unix-domain-socket.hh" #include "lix/libstore/worker-protocol.hh" @@ -76,7 +77,7 @@ ref UDSRemoteStore::openConnection() kj::Promise> UDSRemoteStore::addIndirectRoot(const Path & path) try { - auto conn(getConnection()); + auto conn(TRY_AWAIT(getConnection())); conn->to << WorkerProto::Op::AddIndirectRoot << path; conn.processStderr(); readInt(conn->from); diff --git a/lix/libutil/pool.hh b/lix/libutil/pool.hh index bd8c5366d..c92552f3d 100644 --- a/lix/libutil/pool.hh +++ b/lix/libutil/pool.hh @@ -2,13 +2,19 @@ ///@file #include +#include #include #include #include #include +#include +#include +#include "lix/libutil/async.hh" +#include "lix/libutil/result.hh" #include "lix/libutil/sync.hh" #include "lix/libutil/ref.hh" +#include "lix/libutil/types.hh" namespace nix { @@ -55,12 +61,19 @@ private: size_t inUse = 0; size_t max; std::vector> idle; + std::list>> waiters; + + void notify() + { + for (auto & waiter : waiters) { + waiter->fulfill(); + } + waiters.clear(); + } }; Sync state; - std::condition_variable wakeup; - public: Pool(size_t max = std::numeric_limits::max(), @@ -122,8 +135,8 @@ public: state_->idle.push_back(ref(r)); assert(state_->inUse); state_->inUse--; + state_->notify(); } - pool.wakeup.notify_one(); } R * operator -> () { return &*r; } @@ -132,39 +145,61 @@ public: void markBad() { bad = true; } }; - Handle get() +private: + void getFailed() { - { - auto state_(state.lock()); + auto state_(state.lock()); + state_->inUse--; + state_->notify(); + } - /* If we're over the maximum number of instance, we need - to wait until a slot becomes available. */ - while (state_->idle.empty() && state_->inUse >= state_->max) - state_.wait(wakeup); + // lock lifetimes must always be short, and *NEVER* cross a yield point. + // we ensure this by using explicit continuations instead of coroutines. + kj::Promise>> tryGet() + try { + auto state_(state.lock()); - while (!state_->idle.empty()) { - auto p = state_->idle.back(); - state_->idle.pop_back(); - if (validator(p)) { - state_->inUse++; - return Handle(*this, p); - } + /* If we're over the maximum number of instance, we need + to wait until a slot becomes available. */ + if (state_->idle.empty() && state_->inUse >= state_->max) { + auto pfp = kj::newPromiseAndCrossThreadFulfiller(); + state_->waiters.push_back(std::move(pfp.fulfiller)); + return pfp.promise.then([this] { return tryGet(); }); + } + + while (!state_->idle.empty()) { + auto p = state_->idle.back(); + state_->idle.pop_back(); + if (validator(p)) { + state_->inUse++; + return {Handle(*this, p)}; } + } - state_->inUse++; + state_->inUse++; + return {std::nullopt}; + } catch (...) { + return {result::current_exception()}; + } + +public: + kj::Promise> get() + try { + if (auto existing = LIX_TRY_AWAIT(tryGet())) { + co_return std::move(*existing); } /* We need to create a new instance. Because that might take a while, we don't hold the lock in the meantime. */ try { Handle h(*this, factory()); - return h; + co_return h; } catch (...) { - auto state_(state.lock()); - state_->inUse--; - wakeup.notify_one(); + getFailed(); throw; } + } catch (...) { + co_return result::current_exception(); } size_t count() diff --git a/tests/unit/libutil/pool.cc b/tests/unit/libutil/pool.cc index 5c9d1ff05..2da205d17 100644 --- a/tests/unit/libutil/pool.cc +++ b/tests/unit/libutil/pool.cc @@ -1,5 +1,6 @@ #include "lix/libutil/pool.hh" #include +#include namespace nix { @@ -16,11 +17,20 @@ namespace nix { int num; }; + class PoolTest : public testing::Test + { + public: + kj::EventLoop loop; + kj::WaitScope ws{loop}; + + ~PoolTest() noexcept(true) = default; + }; + /* ---------------------------------------------------------------------------- * Pool * --------------------------------------------------------------------------*/ - TEST(Pool, freshPoolHasZeroCountAndSpecifiedCapacity) { + TEST_F(PoolTest, freshPoolHasZeroCountAndSpecifiedCapacity) { auto isGood = [](const ref & r) { return r->good; }; auto createResource = []() { return make_ref(); }; @@ -30,14 +40,14 @@ namespace nix { ASSERT_EQ(pool.capacity(), 1); } - TEST(Pool, freshPoolCanGetAResource) { + TEST_F(PoolTest, freshPoolCanGetAResource) { auto isGood = [](const ref & r) { return r->good; }; auto createResource = []() { return make_ref(); }; Pool pool = Pool((size_t)1, createResource, isGood); ASSERT_EQ(pool.count(), 0); - TestResource r = *(pool.get()); + TestResource r = *(pool.get().wait(ws).value()); ASSERT_EQ(pool.count(), 1); ASSERT_EQ(pool.capacity(), 1); @@ -45,7 +55,7 @@ namespace nix { ASSERT_EQ(r.good, true); } - TEST(Pool, capacityCanBeIncremented) { + TEST_F(PoolTest, capacityCanBeIncremented) { auto isGood = [](const ref & r) { return r->good; }; auto createResource = []() { return make_ref(); }; @@ -55,7 +65,7 @@ namespace nix { ASSERT_EQ(pool.capacity(), 2); } - TEST(Pool, capacityCanBeDecremented) { + TEST_F(PoolTest, capacityCanBeDecremented) { auto isGood = [](const ref & r) { return r->good; }; auto createResource = []() { return make_ref(); }; @@ -66,7 +76,7 @@ namespace nix { } // Test that the resources we allocate are being reused when they are still good. - TEST(Pool, reuseResource) { + TEST_F(PoolTest, reuseResource) { auto isGood = [](const ref & r) { return true; }; auto createResource = []() { return make_ref(); }; @@ -76,18 +86,18 @@ namespace nix { // as the pool should hand out the same (still) good one again. int counter = -1; { - Pool::Handle h = pool.get(); + Pool::Handle h = pool.get().wait(ws).value(); counter = h->num; } // the first handle goes out of scope { // the second handle should contain the same resource (with the same counter value) - Pool::Handle h = pool.get(); + Pool::Handle h = pool.get().wait(ws).value(); ASSERT_EQ(h->num, counter); } } // Test that the resources we allocate are being thrown away when they are no longer good. - TEST(Pool, badResourceIsNotReused) { + TEST_F(PoolTest, badResourceIsNotReused) { auto isGood = [](const ref & r) { return false; }; auto createResource = []() { return make_ref(); }; @@ -98,14 +108,14 @@ namespace nix { // the first one was returned. int counter = -1; { - Pool::Handle h = pool.get(); + Pool::Handle h = pool.get().wait(ws).value(); counter = h->num; } // the first handle goes out of scope { // the second handle should contain a different resource (with a //different counter value) - Pool::Handle h = pool.get(); + Pool::Handle h = pool.get().wait(ws).value(); ASSERT_NE(h->num, counter); } }