nixos/nix
1
1
Fork 0
mirror of https://github.com/NixOS/nix.git synced 2025-11-16 15:32:43 +01:00
Code Activity

Put worker protocol items inside a WorkerProto struct

Browse source 
See API docs on that struct for why. The pasing as as template argument
doesn't yet happen in that commit, but will instead happen in later
commit.

Also make `WorkerOp` (now `Op`) and enum struct. This led us to catch
that two operations were not handled!

Co-authored-by: Robert Hensing <roberth@users.noreply.github.com>
This commit is contained in:
John Ericson 2023-05-26 11:07:25 -04:00
parent 469d06f9bc
commit 95eae0c002
12 changed files with 327 additions and 273 deletions
Download patch file Download diff file Expand all files Collapse all files

130
src/libstore/remote-store.cc
View file

@ -101,7 +101,7 @@ void RemoteStore::initConnection(Connection & conn)
}
if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 35) {
conn.remoteTrustsUs = WorkerProto<std::optional<TrustedFlag>>::read(*this, conn.from);
conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(*this, conn.from);
} else {
// We don't know the answer; protocol to old.
conn.remoteTrustsUs = std::nullopt;
@ -120,7 +120,7 @@ void RemoteStore::initConnection(Connection & conn)
void RemoteStore::setOptions(Connection & conn)
{
conn.to << wopSetOptions
conn.to << WorkerProto::Op::SetOptions
<< settings.keepFailed
<< settings.keepGoing
<< settings.tryFallback
@ -212,7 +212,7 @@ void RemoteStore::setOptions()
bool RemoteStore::isValidPathUncached(const StorePath & path)
{
auto conn(getConnection());
conn->to << wopIsValidPath << printStorePath(path);
conn->to << WorkerProto::Op::IsValidPath << printStorePath(path);
conn.processStderr();
return readInt(conn->from);
}
@ -227,13 +227,13 @@ StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, Substitute
if (isValidPath(i)) res.insert(i);
return res;
} else {
conn->to << wopQueryValidPaths;
workerProtoWrite(*this, conn->to, paths);
conn->to << WorkerProto::Op::QueryValidPaths;
WorkerProto::write(*this, conn->to, paths);
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 27) {
conn->to << (settings.buildersUseSubstitutes ? 1 : 0);
}
conn.processStderr();
return WorkerProto<StorePathSet>::read(*this, conn->from);
return WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
}
}
@ -241,9 +241,9 @@ StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, Substitute
StorePathSet RemoteStore::queryAllValidPaths()
{
auto conn(getConnection());
conn->to << wopQueryAllValidPaths;
conn->to << WorkerProto::Op::QueryAllValidPaths;
conn.processStderr();
return WorkerProto<StorePathSet>::read(*this, conn->from);
return WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
}
@ -253,16 +253,16 @@ StorePathSet RemoteStore::querySubstitutablePaths(const StorePathSet & paths)
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) {
StorePathSet res;
for (auto & i : paths) {
conn->to << wopHasSubstitutes << printStorePath(i);
conn->to << WorkerProto::Op::HasSubstitutes << printStorePath(i);
conn.processStderr();
if (readInt(conn->from)) res.insert(i);
}
return res;
} else {
conn->to << wopQuerySubstitutablePaths;
workerProtoWrite(*this, conn->to, paths);
conn->to << WorkerProto::Op::QuerySubstitutablePaths;
WorkerProto::write(*this, conn->to, paths);
conn.processStderr();
return WorkerProto<StorePathSet>::read(*this, conn->from);
return WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
}
}
@ -277,14 +277,14 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S
for (auto & i : pathsMap) {
SubstitutablePathInfo info;
conn->to << wopQuerySubstitutablePathInfo << printStorePath(i.first);
conn->to << WorkerProto::Op::QuerySubstitutablePathInfo << printStorePath(i.first);
conn.processStderr();
unsigned int reply = readInt(conn->from);
if (reply == 0) continue;
auto deriver = readString(conn->from);
if (deriver != "")
info.deriver = parseStorePath(deriver);
info.references = WorkerProto<StorePathSet>::read(*this, conn->from);
info.references = WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
info.downloadSize = readLongLong(conn->from);
info.narSize = readLongLong(conn->from);
infos.insert_or_assign(i.first, std::move(info));
@ -292,14 +292,14 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S
} else {
conn->to << wopQuerySubstitutablePathInfos;
conn->to << WorkerProto::Op::QuerySubstitutablePathInfos;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 22) {
StorePathSet paths;
for (auto & path : pathsMap)
paths.insert(path.first);
workerProtoWrite(*this, conn->to, paths);
WorkerProto::write(*this, conn->to, paths);
} else
workerProtoWrite(*this, conn->to, pathsMap);
WorkerProto::write(*this, conn->to, pathsMap);
conn.processStderr();
size_t count = readNum<size_t>(conn->from);
for (size_t n = 0; n < count; n++) {
@ -307,7 +307,7 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S
auto deriver = readString(conn->from);
if (deriver != "")
info.deriver = parseStorePath(deriver);
info.references = WorkerProto<StorePathSet>::read(*this, conn->from);
info.references = WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
info.downloadSize = readLongLong(conn->from);
info.narSize = readLongLong(conn->from);
}
@ -323,7 +323,7 @@ void RemoteStore::queryPathInfoUncached(const StorePath & path,
std::shared_ptr<const ValidPathInfo> info;
{
auto conn(getConnection());
conn->to << wopQueryPathInfo << printStorePath(path);
conn->to << WorkerProto::Op::QueryPathInfo << printStorePath(path);
try {
conn.processStderr();
} catch (Error & e) {
@ -348,9 +348,9 @@ void RemoteStore::queryReferrers(const StorePath & path,
StorePathSet & referrers)
{
auto conn(getConnection());
conn->to << wopQueryReferrers << printStorePath(path);
conn->to << WorkerProto::Op::QueryReferrers << printStorePath(path);
conn.processStderr();
for (auto & i : WorkerProto<StorePathSet>::read(*this, conn->from))
for (auto & i : WorkerProto::Serialise<StorePathSet>::read(*this, conn->from))
referrers.insert(i);
}
@ -358,9 +358,9 @@ void RemoteStore::queryReferrers(const StorePath & path,
StorePathSet RemoteStore::queryValidDerivers(const StorePath & path)
{
auto conn(getConnection());
conn->to << wopQueryValidDerivers << printStorePath(path);
conn->to << WorkerProto::Op::QueryValidDerivers << printStorePath(path);
conn.processStderr();
return WorkerProto<StorePathSet>::read(*this, conn->from);
return WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
}
@ -370,9 +370,9 @@ StorePathSet RemoteStore::queryDerivationOutputs(const StorePath & path)
return Store::queryDerivationOutputs(path);
}
auto conn(getConnection());
conn->to << wopQueryDerivationOutputs << printStorePath(path);
conn->to << WorkerProto::Op::QueryDerivationOutputs << printStorePath(path);
conn.processStderr();
return WorkerProto<StorePathSet>::read(*this, conn->from);
return WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
}
@ -380,9 +380,9 @@ std::map<std::string, std::optional<StorePath>> RemoteStore::queryPartialDerivat
{
if (GET_PROTOCOL_MINOR(getProtocol()) >= 0x16) {
auto conn(getConnection());
conn->to << wopQueryDerivationOutputMap << printStorePath(path);
conn->to << WorkerProto::Op::QueryDerivationOutputMap << printStorePath(path);
conn.processStderr();
return WorkerProto<std::map<std::string, std::optional<StorePath>>>::read(*this, conn->from);
return WorkerProto::Serialise<std::map<std::string, std::optional<StorePath>>>::read(*this, conn->from);
} else {
// Fallback for old daemon versions.
// For floating-CA derivations (and their co-dependencies) this is an
@ -403,7 +403,7 @@ std::map<std::string, std::optional<StorePath>> RemoteStore::queryPartialDerivat
std::optional<StorePath> RemoteStore::queryPathFromHashPart(const std::string & hashPart)
{
auto conn(getConnection());
conn->to << wopQueryPathFromHashPart << hashPart;
conn->to << WorkerProto::Op::QueryPathFromHashPart << hashPart;
conn.processStderr();
Path path = readString(conn->from);
if (path.empty()) return {};
@ -425,10 +425,10 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 25) {
conn->to
<< wopAddToStore
<< WorkerProto::Op::AddToStore
<< name
<< caMethod.render(hashType);
workerProtoWrite(*this, conn->to, references);
WorkerProto::write(*this, conn->to, references);
conn->to << repair;
// The dump source may invoke the store, so we need to make some room.
@ -452,13 +452,13 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
throw UnimplementedError("When adding text-hashed data called '%s', only SHA-256 is supported but '%s' was given",
name, printHashType(hashType));
std::string s = dump.drain();
conn->to << wopAddTextToStore << name << s;
workerProtoWrite(*this, conn->to, references);
conn->to << WorkerProto::Op::AddTextToStore << name << s;
WorkerProto::write(*this, conn->to, references);
conn.processStderr();
},
[&](const FileIngestionMethod & fim) -> void {
conn->to
<< wopAddToStore
<< WorkerProto::Op::AddToStore
<< name
<< ((hashType == htSHA256 && fim == FileIngestionMethod::Recursive) ? 0 : 1) /* backwards compatibility hack */
<< (fim == FileIngestionMethod::Recursive ? 1 : 0)
@ -510,7 +510,7 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
auto conn(getConnection());
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 18) {
conn->to << wopImportPaths;
conn->to << WorkerProto::Op::ImportPaths;
auto source2 = sinkToSource([&](Sink & sink) {
sink << 1 // == path follows
@ -519,7 +519,7 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
sink
<< exportMagic
<< printStorePath(info.path);
workerProtoWrite(*this, sink, info.references);
WorkerProto::write(*this, sink, info.references);
sink
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< 0 // == no legacy signature
@ -529,16 +529,16 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
conn.processStderr(0, source2.get());
auto importedPaths = WorkerProto<StorePathSet>::read(*this, conn->from);
auto importedPaths = WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
assert(importedPaths.size() <= 1);
}
else {
conn->to << wopAddToStoreNar
conn->to << WorkerProto::Op::AddToStoreNar
<< printStorePath(info.path)
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< info.narHash.to_string(Base16, false);
workerProtoWrite(*this, conn->to, info.references);
WorkerProto::write(*this, conn->to, info.references);
conn->to << info.registrationTime << info.narSize
<< info.ultimate << info.sigs << renderContentAddress(info.ca)
<< repair << !checkSigs;
@ -582,7 +582,7 @@ void RemoteStore::addMultipleToStore(
if (GET_PROTOCOL_MINOR(getConnection()->daemonVersion) >= 32) {
auto conn(getConnection());
conn->to
<< wopAddMultipleToStore
<< WorkerProto::Op::AddMultipleToStore
<< repair
<< !checkSigs;
conn.withFramedSink([&](Sink & sink) {
@ -606,12 +606,12 @@ StorePath RemoteStore::addTextToStore(
void RemoteStore::registerDrvOutput(const Realisation & info)
{
auto conn(getConnection());
conn->to << wopRegisterDrvOutput;
conn->to << WorkerProto::Op::RegisterDrvOutput;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) {
conn->to << info.id.to_string();
conn->to << std::string(info.outPath.to_string());
} else {
workerProtoWrite(*this, conn->to, info);
WorkerProto::write(*this, conn->to, info);
}
conn.processStderr();
}
@ -627,19 +627,19 @@ void RemoteStore::queryRealisationUncached(const DrvOutput & id,
return callback(nullptr);
}
conn->to << wopQueryRealisation;
conn->to << WorkerProto::Op::QueryRealisation;
conn->to << id.to_string();
conn.processStderr();
auto real = [&]() -> std::shared_ptr<const Realisation> {
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) {
auto outPaths = WorkerProto<std::set<StorePath>>::read(
auto outPaths = WorkerProto::Serialise<std::set<StorePath>>::read(
*this, conn->from);
if (outPaths.empty())
return nullptr;
return std::make_shared<const Realisation>(Realisation { .id = id, .outPath = *outPaths.begin() });
} else {
auto realisations = WorkerProto<std::set<Realisation>>::read(
auto realisations = WorkerProto::Serialise<std::set<Realisation>>::read(
*this, conn->from);
if (realisations.empty())
return nullptr;
@ -654,7 +654,7 @@ void RemoteStore::queryRealisationUncached(const DrvOutput & id,
static void writeDerivedPaths(RemoteStore & store, ConnectionHandle & conn, const std::vector<DerivedPath> & reqs)
{
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 30) {
workerProtoWrite(store, conn->to, reqs);
WorkerProto::write(store, conn->to, reqs);
} else {
Strings ss;
for (auto & p : reqs) {
@ -695,7 +695,7 @@ void RemoteStore::buildPaths(const std::vector<DerivedPath> & drvPaths, BuildMod
copyDrvsFromEvalStore(drvPaths, evalStore);
auto conn(getConnection());
conn->to << wopBuildPaths;
conn->to << WorkerProto::Op::BuildPaths;
assert(GET_PROTOCOL_MINOR(conn->daemonVersion) >= 13);
writeDerivedPaths(*this, conn, drvPaths);
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 15)
@ -720,11 +720,11 @@ std::vector<KeyedBuildResult> RemoteStore::buildPathsWithResults(
auto & conn = *conn_;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) {
conn->to << wopBuildPathsWithResults;
conn->to << WorkerProto::Op::BuildPathsWithResults;
writeDerivedPaths(*this, conn, paths);
conn->to << buildMode;
conn.processStderr();
return WorkerProto<std::vector<KeyedBuildResult>>::read(*this, conn->from);
return WorkerProto::Serialise<std::vector<KeyedBuildResult>>::read(*this, conn->from);
} else {
// Avoid deadlock.
conn_.reset();
@ -796,7 +796,7 @@ BuildResult RemoteStore::buildDerivation(const StorePath & drvPath, const BasicD
BuildMode buildMode)
{
auto conn(getConnection());
conn->to << wopBuildDerivation << printStorePath(drvPath);
conn->to << WorkerProto::Op::BuildDerivation << printStorePath(drvPath);
writeDerivation(conn->to, *this, drv);
conn->to << buildMode;
conn.processStderr();
@ -807,7 +807,7 @@ BuildResult RemoteStore::buildDerivation(const StorePath & drvPath, const BasicD
conn->from >> res.timesBuilt >> res.isNonDeterministic >> res.startTime >> res.stopTime;
}
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 28) {
auto builtOutputs = WorkerProto<DrvOutputs>::read(*this, conn->from);
auto builtOutputs = WorkerProto::Serialise<DrvOutputs>::read(*this, conn->from);
for (auto && [output, realisation] : builtOutputs)
res.builtOutputs.insert_or_assign(
std::move(output.outputName),
@ -820,7 +820,7 @@ BuildResult RemoteStore::buildDerivation(const StorePath & drvPath, const BasicD
void RemoteStore::ensurePath(const StorePath & path)
{
auto conn(getConnection());
conn->to << wopEnsurePath << printStorePath(path);
conn->to << WorkerProto::Op::EnsurePath << printStorePath(path);
conn.processStderr();
readInt(conn->from);
}
@ -829,7 +829,7 @@ void RemoteStore::ensurePath(const StorePath & path)
void RemoteStore::addTempRoot(const StorePath & path)
{
auto conn(getConnection());
conn->to << wopAddTempRoot << printStorePath(path);
conn->to << WorkerProto::Op::AddTempRoot << printStorePath(path);
conn.processStderr();
readInt(conn->from);
}
@ -838,7 +838,7 @@ void RemoteStore::addTempRoot(const StorePath & path)
void RemoteStore::addIndirectRoot(const Path & path)
{
auto conn(getConnection());
conn->to << wopAddIndirectRoot << path;
conn->to << WorkerProto::Op::AddIndirectRoot << path;
conn.processStderr();
readInt(conn->from);
}
@ -847,7 +847,7 @@ void RemoteStore::addIndirectRoot(const Path & path)
Roots RemoteStore::findRoots(bool censor)
{
auto conn(getConnection());
conn->to << wopFindRoots;
conn->to << WorkerProto::Op::FindRoots;
conn.processStderr();
size_t count = readNum<size_t>(conn->from);
Roots result;
@ -865,8 +865,8 @@ void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results)
auto conn(getConnection());
conn->to
<< wopCollectGarbage << options.action;
workerProtoWrite(*this, conn->to, options.pathsToDelete);
<< WorkerProto::Op::CollectGarbage << options.action;
WorkerProto::write(*this, conn->to, options.pathsToDelete);
conn->to << options.ignoreLiveness
<< options.maxFreed
/* removed options */
@ -888,7 +888,7 @@ void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results)
void RemoteStore::optimiseStore()
{
auto conn(getConnection());
conn->to << wopOptimiseStore;
conn->to << WorkerProto::Op::OptimiseStore;
conn.processStderr();
readInt(conn->from);
}
@ -897,7 +897,7 @@ void RemoteStore::optimiseStore()
bool RemoteStore::verifyStore(bool checkContents, RepairFlag repair)
{
auto conn(getConnection());
conn->to << wopVerifyStore << checkContents << repair;
conn->to << WorkerProto::Op::VerifyStore << checkContents << repair;
conn.processStderr();
return readInt(conn->from);
}
@ -906,7 +906,7 @@ bool RemoteStore::verifyStore(bool checkContents, RepairFlag repair)
void RemoteStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
{
auto conn(getConnection());
conn->to << wopAddSignatures << printStorePath(storePath) << sigs;
conn->to << WorkerProto::Op::AddSignatures << printStorePath(storePath) << sigs;
conn.processStderr();
readInt(conn->from);
}
@ -922,12 +922,12 @@ void RemoteStore::queryMissing(const std::vector<DerivedPath> & targets,
// Don't hold the connection handle in the fallback case
// to prevent a deadlock.
goto fallback;
conn->to << wopQueryMissing;
conn->to << WorkerProto::Op::QueryMissing;
writeDerivedPaths(*this, conn, targets);
conn.processStderr();
willBuild = WorkerProto<StorePathSet>::read(*this, conn->from);
willSubstitute = WorkerProto<StorePathSet>::read(*this, conn->from);
unknown = WorkerProto<StorePathSet>::read(*this, conn->from);
willBuild = WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
willSubstitute = WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
unknown = WorkerProto::Serialise<StorePathSet>::read(*this, conn->from);
conn->from >> downloadSize >> narSize;
return;
}
@ -941,7 +941,7 @@ void RemoteStore::queryMissing(const std::vector<DerivedPath> & targets,
void RemoteStore::addBuildLog(const StorePath & drvPath, std::string_view log)
{
auto conn(getConnection());
conn->to << wopAddBuildLog << drvPath.to_string();
conn->to << WorkerProto::Op::AddBuildLog << drvPath.to_string();
StringSource source(log);
conn.withFramedSink([&](Sink & sink) {
source.drainInto(sink);
@ -993,7 +993,7 @@ RemoteStore::Connection::~Connection()
void RemoteStore::narFromPath(const StorePath & path, Sink & sink)
{
auto conn(connections->get());
conn->to << wopNarFromPath << printStorePath(path);
conn->to << WorkerProto::Op::NarFromPath << printStorePath(path);
conn->processStderr();
copyNAR(conn->from, sink);
}