From 1a2247560df37141a439fa889502c977e8b2a98f Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Wed, 11 Jun 2025 17:22:33 +0200 Subject: [PATCH] libstore: encapsulate reading simple command results much the same as the previous change, but on the receiving side. Change-Id: I9f8a156a9d8fccaf91347e34a5b6baf301df5800 --- lix/libstore/remote-store-connection.hh | 13 +++- lix/libstore/remote-store.cc | 95 +++++++++++-------------- lix/libstore/uds-remote-store.cc | 3 +- 3 files changed, 53 insertions(+), 58 deletions(-) diff --git a/lix/libstore/remote-store-connection.hh b/lix/libstore/remote-store-connection.hh index af5be1fff..dd255c8b3 100644 --- a/lix/libstore/remote-store-connection.hh +++ b/lix/libstore/remote-store-connection.hh @@ -3,8 +3,11 @@ #include "lix/libstore/remote-store.hh" #include "lix/libstore/worker-protocol.hh" +#include "lix/libstore/worker-protocol-impl.hh" #include "lix/libutil/pool.hh" #include "lix/libutil/result.hh" +#include "lix/libutil/serialise.hh" +#include namespace nix { @@ -134,12 +137,16 @@ struct RemoteStore::ConnectionHandle kj::Promise> withFramedSinkAsync(std::function>(Sink & sink)> fun); - template - kj::Promise> sendCommand(Args &&... args) + template + kj::Promise> sendCommand(Args &&... args) try { ((handle->to << std::forward(args)), ...); processStderr(); - co_return result::success(); + if constexpr (std::is_void_v) { + co_return result::success(); + } else { + co_return WorkerProto::Serialise::read(*handle); + } } catch (...) { co_return result::current_exception(); } diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index df2ee2155..91b181f3c 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -195,8 +195,9 @@ try { kj::Promise> RemoteStore::isValidPathUncached(const StorePath & path) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::IsValidPath, printStorePath(path))); - co_return readInt(conn->from); + co_return TRY_AWAIT( + conn.sendCommand(WorkerProto::Op::IsValidPath, printStorePath(path)) + ); } catch (...) { co_return result::current_exception(); } @@ -206,10 +207,9 @@ kj ::Promise> RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand( + co_return TRY_AWAIT(conn.sendCommand( WorkerProto::Op::QueryValidPaths, WorkerProto::write(*conn, paths), maybeSubstitute )); - co_return WorkerProto::Serialise::read(*conn); } catch (...) { co_return result::current_exception(); } @@ -218,8 +218,7 @@ try { kj::Promise> RemoteStore::queryAllValidPaths() try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryAllValidPaths)); - co_return WorkerProto::Serialise::read(*conn); + co_return TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryAllValidPaths)); } catch (...) { co_return result::current_exception(); } @@ -228,10 +227,9 @@ try { kj::Promise> RemoteStore::querySubstitutablePaths(const StorePathSet & paths) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT( - conn.sendCommand(WorkerProto::Op::QuerySubstitutablePaths, WorkerProto::write(*conn, paths)) - ); - co_return WorkerProto::Serialise::read(*conn); + co_return TRY_AWAIT(conn.sendCommand( + WorkerProto::Op::QuerySubstitutablePaths, WorkerProto::write(*conn, paths) + )); } catch (...) { co_return result::current_exception(); } @@ -243,10 +241,9 @@ try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand( + infos = TRY_AWAIT(conn.sendCommand( WorkerProto::Op::QuerySubstitutablePathInfos, WorkerProto::write(*conn, pathsMap) )); - infos = WorkerProto::Serialise::read(*conn); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -258,7 +255,9 @@ RemoteStore::queryPathInfoUncached(const StorePath & path) try { auto conn(TRY_AWAIT(getConnection())); try { - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryPathInfo, printStorePath(path))); + auto valid = + TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryPathInfo, printStorePath(path))); + if (!valid) co_return result::success(nullptr); } catch (Error & e) { // Ugly backwards compatibility hack. TODO(fj#325): remove. if (e.msg().find("is not valid") != std::string::npos) @@ -266,9 +265,6 @@ try { throw; } - bool valid; conn->from >> valid; - if (!valid) co_return result::success(nullptr); - co_return std::make_shared( StorePath{path}, WorkerProto::Serialise::read(*conn)); @@ -293,8 +289,9 @@ try { kj::Promise> RemoteStore::queryValidDerivers(const StorePath & path) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryValidDerivers, printStorePath(path))); - co_return WorkerProto::Serialise::read(*conn); + co_return TRY_AWAIT( + conn.sendCommand(WorkerProto::Op::QueryValidDerivers, printStorePath(path)) + ); } catch (...) { co_return result::current_exception(); } @@ -305,10 +302,9 @@ RemoteStore::queryDerivationOutputMap(const StorePath & path, Store * evalStore_ try { if (!evalStore_) { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryDerivationOutputMap, printStorePath(path))); - auto tmp = WorkerProto::Serialise>>::read( - *conn - ); + auto tmp = TRY_AWAIT(conn.sendCommand>>( + WorkerProto::Op::QueryDerivationOutputMap, printStorePath(path) + )); std::map result; for (auto & [name, outPath] : tmp) { if (!outPath) { @@ -339,8 +335,8 @@ kj::Promise>> RemoteStore::queryPathFromHashPart(const std::string & hashPart) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryPathFromHashPart, hashPart)); - Path path = readString(conn->from); + Path path = + TRY_AWAIT(conn.sendCommand(WorkerProto::Op::QueryPathFromHashPart, hashPart)); if (path.empty()) co_return std::nullopt; co_return parseStorePath(path); } catch (...) { @@ -502,10 +498,9 @@ try { TRY_AWAIT(copyDrvsFromEvalStore(drvPaths, evalStore)); auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand( + TRY_AWAIT(conn.sendCommand( WorkerProto::Op::BuildPaths, WorkerProto::write(*conn, drvPaths), buildMode )); - readInt(conn->from); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -520,10 +515,9 @@ try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand( + co_return TRY_AWAIT(conn.sendCommand>( WorkerProto::Op::BuildPathsWithResults, WorkerProto::write(*conn, paths), buildMode )); - co_return WorkerProto::Serialise>::read(*conn); } catch (...) { co_return result::current_exception(); } @@ -533,13 +527,12 @@ kj ::Promise> RemoteStore::buildDerivation( ) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand( + co_return TRY_AWAIT(conn.sendCommand( WorkerProto::Op::BuildDerivation, printStorePath(drvPath), serializeDerivation(*this, drv), buildMode )); - co_return WorkerProto::Serialise::read(*conn); } catch (...) { co_return result::current_exception(); } @@ -548,8 +541,7 @@ try { kj::Promise> RemoteStore::ensurePath(const StorePath & path) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::EnsurePath, printStorePath(path))); - readInt(conn->from); + TRY_AWAIT(conn.sendCommand(WorkerProto::Op::EnsurePath, printStorePath(path))); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -559,8 +551,7 @@ try { kj::Promise> RemoteStore::addTempRoot(const StorePath & path) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::AddTempRoot, printStorePath(path))); - readInt(conn->from); + TRY_AWAIT(conn.sendCommand(WorkerProto::Op::AddTempRoot, printStorePath(path))); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -570,10 +561,10 @@ try { kj::Promise> RemoteStore::findRoots(bool censor) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::FindRoots)); - auto roots = - WorkerProto::Serialise>>::read(*conn); + auto roots = TRY_AWAIT(conn.sendCommand>>( + WorkerProto::Op::FindRoots + )); Roots result; for (auto & [link, target] : roots) { result[std::move(target)].emplace(link); @@ -589,7 +580,9 @@ RemoteStore::collectGarbage(const GCOptions & options, GCResults & results) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand( + using ResultT = std::tuple; + + std::tie(results.paths, results.bytesFreed, std::ignore) = TRY_AWAIT(conn.sendCommand( WorkerProto::Op::CollectGarbage, options.action, WorkerProto::write(*conn, options.pathsToDelete), @@ -599,10 +592,6 @@ try { 0, 0, 0 )); - results.paths = readStrings(conn->from); - results.bytesFreed = readLongLong(conn->from); - readLongLong(conn->from); // obsolete - { auto state_(co_await Store::state.lock()); state_->pathInfoCache.clear(); @@ -616,8 +605,7 @@ try { kj::Promise> RemoteStore::optimiseStore() try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::OptimiseStore)); - readInt(conn->from); + TRY_AWAIT(conn.sendCommand(WorkerProto::Op::OptimiseStore)); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -627,8 +615,9 @@ try { kj::Promise> RemoteStore::verifyStore(bool checkContents, RepairFlag repair) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::VerifyStore, checkContents, repair)); - co_return readInt(conn->from); + co_return TRY_AWAIT( + conn.sendCommand(WorkerProto::Op::VerifyStore, checkContents, repair) + ); } catch (...) { co_return result::current_exception(); } @@ -638,8 +627,9 @@ kj::Promise> RemoteStore::addSignatures(const StorePath & storePath, const StringSet & sigs) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::AddSignatures, printStorePath(storePath), sigs)); - readInt(conn->from); + TRY_AWAIT( + conn.sendCommand(WorkerProto::Op::AddSignatures, printStorePath(storePath), sigs) + ); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -651,11 +641,10 @@ kj::Promise> RemoteStore::queryMissing(const std::vector::read(*conn); - willSubstitute = WorkerProto::Serialise::read(*conn); - unknown = WorkerProto::Serialise::read(*conn); - conn->from >> downloadSize >> narSize; + using ResultT = std::tuple; + std::tie(willBuild, willSubstitute, unknown, downloadSize, narSize) = TRY_AWAIT( + conn.sendCommand(WorkerProto::Op::QueryMissing, WorkerProto::write(*conn, targets)) + ); co_return result::success(); } catch (...) { co_return result::current_exception(); diff --git a/lix/libstore/uds-remote-store.cc b/lix/libstore/uds-remote-store.cc index 0c81065f1..142ccb0c5 100644 --- a/lix/libstore/uds-remote-store.cc +++ b/lix/libstore/uds-remote-store.cc @@ -72,8 +72,7 @@ ref UDSRemoteStore::openConnection() kj::Promise> UDSRemoteStore::addIndirectRoot(const Path & path) try { auto conn(TRY_AWAIT(getConnection())); - TRY_AWAIT(conn.sendCommand(WorkerProto::Op::AddIndirectRoot, path)); - readInt(conn->from); + TRY_AWAIT(conn.sendCommand(WorkerProto::Op::AddIndirectRoot, path)); co_return result::success(); } catch (...) { co_return result::current_exception();