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

libutil/libstore: asyncify Pool

Browse Source 
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
This commit is contained in:
eldritch horrors
2025-03-05 23:07:20 +01:00
parent 2a1a78c025
commit 4e266a25d6
7 changed files with 163 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/legacy-ssh-store.cc
+11 -11
View File
@@ -163,7 +163,7 @@ struct LegacySSHStore final : public Store
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
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<Result<box_ptr<Source>>> 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<BuildResult>::read(*this, *conn)};
co_return ServeProto::Serialise<BuildResult>::read(*this, *conn);
} catch (...) {
return {result::current_exception()};
co_return result::current_exception();
}
kj ::Promise<Result<void>> 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<Result<StorePathSet>> 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<Result<void>> 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<Result<unsigned int>> getProtocol() override
try {
auto conn(connections->get());
auto conn(TRY_AWAIT(connections->get()));
co_return conn->remoteVersion;
} catch (...) {
co_return result::current_exception();
lix/libstore/local-store.cc
+14 -14
View File
@@ -792,7 +792,7 @@ try {
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
TRY_AWAIT(retrySQLite([&]() -> kj::Promise<Result<void>> {
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<Result<std::shared_ptr<const ValidPathInfo>>> {
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<Result<bool>> {
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<Result<StorePathSet>> {
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<Result<void>> {
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<Result<StorePathSet>> {
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<Result<std::map<std::string, std::optional<StorePath>>>> {
try {
auto state = dbPool.get();
auto state = TRY_AWAIT(dbPool.get());
std::map<std::string, std::optional<StorePath>> 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<Result<std::optional<StorePath>>> {
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<Result<void>> {
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<Result<void>> {
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<Result<void>> {
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<Result<std::optional<const Realisation>>> {
try {
auto state = dbPool.get();
auto state = TRY_AWAIT(dbPool.get());
co_return queryRealisation_(*state, id);
} catch (...) {
co_return result::current_exception();
lix/libstore/remote-store.cc
+57 -54
View File
@@ -190,14 +190,16 @@ void RemoteStore::ConnectionHandle::processStderr(Sink * sink, Source * source,
}
RemoteStore::ConnectionHandle RemoteStore::getConnection()
{
return ConnectionHandle(connections->get());
kj::Promise<Result<RemoteStore::ConnectionHandle>> RemoteStore::getConnection()
try {
co_return ConnectionHandle(TRY_AWAIT(connections->get()));
} catch (...) {
co_return result::current_exception();
}
kj::Promise<Result<void>> 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<Result<bool>> 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<Result<StorePathSet>>
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<Result<StorePathSet>> RemoteStore::queryAllValidPaths()
try {
auto conn(getConnection());
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryAllValidPaths;
conn.processStderr();
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
@@ -243,21 +245,21 @@ try {
kj::Promise<Result<StorePathSet>> 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<StorePathSet>::read(*this, *conn)};
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
} catch (...) {
return {result::current_exception()};
co_return result::current_exception();
}
kj::Promise<Result<void>> 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<Result<std::shared_ptr<const ValidPathInfo>>>
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<Result<void>> 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<StorePathSet>::read(*this, *conn))
@@ -327,7 +329,7 @@ try {
kj::Promise<Result<StorePathSet>> 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<StorePathSet>::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<StorePathSet>::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<std::map<std::string, std::optional<StorePath>>>::read(
@@ -395,7 +397,7 @@ try {
kj::Promise<Result<std::optional<StorePath>>>
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<Result<ref<const ValidPathInfo>>> RemoteStore::addCAToStore(
const StorePathSet & references,
RepairFlag repair)
try {
std::optional<ConnectionHandle> conn_(getConnection());
std::optional<ConnectionHandle> conn_(TRY_AWAIT(getConnection()));
auto & conn = *conn_;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 25) {
@@ -529,7 +531,7 @@ kj::Promise<Result<void>> 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<Result<void>> 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<Result<void>> 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<Result<std::shared_ptr<const Realisation>>>
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<Result<void>> 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<Result<std::vector<KeyedBuildResult>>> RemoteStore::buildPathsWithRe
try {
TRY_AWAIT(copyDrvsFromEvalStore(paths, evalStore));
std::optional<ConnectionHandle> conn_(getConnection());
std::optional<ConnectionHandle> conn_(TRY_AWAIT(getConnection()));
auto & conn = *conn_;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) {
@@ -809,44 +811,44 @@ kj ::Promise<Result<BuildResult>> 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<BuildResult>::read(*this, *conn)};
co_return WorkerProto::Serialise<BuildResult>::read(*this, *conn);
} catch (...) {
return {result::current_exception()};
co_return result::current_exception();
}
kj::Promise<Result<void>> 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<Result<void>> 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<Result<Roots>> RemoteStore::findRoots(bool censor)
try {
auto conn(getConnection());
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::FindRoots;
conn.processStderr();
size_t count = readNum<size_t>(conn->from);
@@ -865,7 +867,7 @@ try {
kj::Promise<Result<void>>
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<Result<void>> 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<Result<bool>> 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<Result<void>>
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<Result<void>> RemoteStore::queryMissing(const std::vector<DerivedPat
StorePathSet & willBuild, StorePathSet & willSubstitute, StorePathSet & unknown,
uint64_t & downloadSize, uint64_t & narSize)
try {
auto conn(getConnection());
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryMissing;
conn->to << WorkerProto::write(*this, *conn, targets);
conn.processStderr();
@@ -947,7 +949,7 @@ try {
kj::Promise<Result<void>> 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<Result<std::optional<std::string>>> 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<Result<void>> 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<Result<unsigned int>> 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<Result<std::optional<TrustedFlag>>> 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<Result<box_ptr<Source>>> 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<GeneratorSource>([](auto conn) -> WireFormatGenerator {
co_yield copyNAR(conn->from);
}(std::move(conn)));
} catch (...) {
co_return result::current_exception();}
co_return result::current_exception();
}
ref<FSAccessor> RemoteStore::getFSAccessor()
lix/libstore/remote-store.hh
+1 -1
View File
@@ -194,7 +194,7 @@ protected:
struct ConnectionHandle;
ConnectionHandle getConnection();
kj::Promise<Result<ConnectionHandle>> getConnection();
friend struct ConnectionHandle;
lix/libstore/uds-remote-store.cc
+2 -1
View File
@@ -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<RemoteStore::Connection> UDSRemoteStore::openConnection()
kj::Promise<Result<void>> 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);
lix/libutil/pool.hh
+57 -22
View File
@@ -2,13 +2,19 @@
///@file
#include <functional>
#include <kj/async.h>
#include <limits>
#include <list>
#include <memory>
#include <cassert>
#include <optional>
#include <vector>
#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<ref<R>> idle;
std::list<kj::Own<kj::CrossThreadPromiseFulfiller<void>>> waiters;
void notify()
{
for (auto & waiter : waiters) {
waiter->fulfill();
}
waiters.clear();
}
};
Sync<State> state;
std::condition_variable wakeup;
public:
Pool(size_t max = std::numeric_limits<size_t>::max(),
@@ -122,8 +135,8 @@ public:
state_->idle.push_back(ref<R>(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<Result<std::optional<Handle>>> 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<void>();
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<Result<Handle>> 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()
tests/unit/libutil/pool.cc
+21 -11
View File
@@ -1,5 +1,6 @@
#include "lix/libutil/pool.hh"
#include <gtest/gtest.h>
#include <kj/async.h>
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<TestResource> & r) { return r->good; };
auto createResource = []() { return make_ref<TestResource>(); };
@@ -30,14 +40,14 @@ namespace nix {
ASSERT_EQ(pool.capacity(), 1);
}
TEST(Pool, freshPoolCanGetAResource) {
TEST_F(PoolTest, freshPoolCanGetAResource) {
auto isGood = [](const ref<TestResource> & r) { return r->good; };
auto createResource = []() { return make_ref<TestResource>(); };
Pool<TestResource> pool = Pool<TestResource>((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<TestResource> & r) { return r->good; };
auto createResource = []() { return make_ref<TestResource>(); };
@@ -55,7 +65,7 @@ namespace nix {
ASSERT_EQ(pool.capacity(), 2);
}
TEST(Pool, capacityCanBeDecremented) {
TEST_F(PoolTest, capacityCanBeDecremented) {
auto isGood = [](const ref<TestResource> & r) { return r->good; };
auto createResource = []() { return make_ref<TestResource>(); };
@@ -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<TestResource> & r) { return true; };
auto createResource = []() { return make_ref<TestResource>(); };
@@ -76,18 +86,18 @@ namespace nix {
// as the pool should hand out the same (still) good one again.
int counter = -1;
{
Pool<TestResource>::Handle h = pool.get();
Pool<TestResource>::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<TestResource>::Handle h = pool.get();
Pool<TestResource>::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<TestResource> & r) { return false; };
auto createResource = []() { return make_ref<TestResource>(); };
@@ -98,14 +108,14 @@ namespace nix {
// the first one was returned.
int counter = -1;
{
Pool<TestResource>::Handle h = pool.get();
Pool<TestResource>::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<TestResource>::Handle h = pool.get();
Pool<TestResource>::Handle h = pool.get().wait(ws).value();
ASSERT_NE(h->num, counter);
}
}