From 469d06f9bc9b2543f48d8e633e47f9344fba35ab Mon Sep 17 00:00:00 2001 From: John Ericson Date: Tue, 8 Mar 2022 21:53:26 +0000 Subject: [PATCH 1/5] Split out worker protocol template definitions from declarations This is generally a fine practice: Putting implementations in headers makes them harder to read and slows compilation. Unfortunately it is necessary for templates, but we can ameliorate that by putting them in a separate header. Only files which need to instantiate those templates will need to include the header with the implementation; the rest can just include the declaration. This is now documenting in the contributing guide. Also, it just happens that these polymorphic serializers are the protocol agnostic ones. (Worker and serve protocol have the same logic for these container types.) This means by doing this general template cleanup, we are also getting a head start on better indicating which code is protocol-specific and which code is shared between protocols. --- doc/manual/src/SUMMARY.md.in | 1 + doc/manual/src/contributing/cxx.md | 28 ++++++++ src/libstore/build/derivation-goal.cc | 1 + src/libstore/build/local-derivation-goal.cc | 1 + src/libstore/daemon.cc | 1 + src/libstore/derivations.cc | 1 + src/libstore/export-import.cc | 1 + src/libstore/legacy-ssh-store.cc | 1 + src/libstore/path-info.cc | 1 + src/libstore/remote-store.cc | 1 + src/libstore/worker-protocol-impl.hh | 78 +++++++++++++++++++++ src/libstore/worker-protocol.cc | 1 + src/libstore/worker-protocol.hh | 63 ----------------- src/libutil/config-impl.hh | 3 + src/nix-store/nix-store.cc | 1 + 15 files changed, 120 insertions(+), 63 deletions(-) create mode 100644 doc/manual/src/contributing/cxx.md create mode 100644 src/libstore/worker-protocol-impl.hh diff --git a/doc/manual/src/SUMMARY.md.in b/doc/manual/src/SUMMARY.md.in index 13d2e4d15..bbb603713 100644 --- a/doc/manual/src/SUMMARY.md.in +++ b/doc/manual/src/SUMMARY.md.in @@ -106,6 +106,7 @@ - [Hacking](contributing/hacking.md) - [Experimental Features](contributing/experimental-features.md) - [CLI guideline](contributing/cli-guideline.md) + - [C++ style guide](contributing/cxx.md) - [Release Notes](release-notes/release-notes.md) - [Release X.Y (202?-??-??)](release-notes/rl-next.md) - [Release 2.16 (2023-05-31)](release-notes/rl-2.16.md) diff --git a/doc/manual/src/contributing/cxx.md b/doc/manual/src/contributing/cxx.md new file mode 100644 index 000000000..ff9297158 --- /dev/null +++ b/doc/manual/src/contributing/cxx.md @@ -0,0 +1,28 @@ +# C++ style guide + +Some miscellaneous notes on how we write C++. +Formatting we hope to eventually normalize automatically, so this section is free to just discuss higher-level concerns. + +## The `*-impl.hh` pattern + +Let's start with some background info first. +Headers, are supposed to contain declarations, not definitions. +This allows us to change a definition without changing the declaration, and have a very small rebuild during development. +Templates, however, need to be specialized to use-sites. +Absent fancier techniques, templates require that the definition, not just mere declaration, must be available at use-sites in order to make that specialization on the fly as part of compiling those use-sites. +Making definitions available like that means putting them in headers, but that is unfortunately means we get all the extra rebuilds we want to avoid by just putting declarations there as described above. + +The `*-impl.hh` pattern is a ham-fisted partial solution to this problem. +It constitutes: + +- Declaring items only in the main `foo.hh`, including templates. + +- Putting template definitions in a companion `foo-impl.hh` header. + +Most C++ developers would accompany this by having `foo.hh` include `foo-impl.hh`, to ensure any file getting the template declarations also got the template definitions. +But we've found not doing this has some benefits and fewer than imagined downsides. +The fact remains that headers are rarely as minimal as they could be; +there is often code that needs declarations from the headers but not the templates within them. +With our pattern where `foo.hh` doesn't include `foo-impl.hh`, that means they can just include `foo.hh` +Code that needs both just includes `foo.hh` and `foo-impl.hh`. +This does make linking error possible where something forgets to include `foo-impl.hh` that needs it, but those are build-time only as easy to fix. diff --git a/src/libstore/build/derivation-goal.cc b/src/libstore/build/derivation-goal.cc index df7d21e54..e02000c8b 100644 --- a/src/libstore/build/derivation-goal.cc +++ b/src/libstore/build/derivation-goal.cc @@ -9,6 +9,7 @@ #include "archive.hh" #include "compression.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "topo-sort.hh" #include "callback.hh" #include "local-store.hh" // TODO remove, along with remaining downcasts diff --git a/src/libstore/build/local-derivation-goal.cc b/src/libstore/build/local-derivation-goal.cc index 7f87cdf55..6e06f6061 100644 --- a/src/libstore/build/local-derivation-goal.cc +++ b/src/libstore/build/local-derivation-goal.cc @@ -11,6 +11,7 @@ #include "compression.hh" #include "daemon.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "topo-sort.hh" #include "callback.hh" #include "json-utils.hh" diff --git a/src/libstore/daemon.cc b/src/libstore/daemon.cc index b6dd83684..54b089b30 100644 --- a/src/libstore/daemon.cc +++ b/src/libstore/daemon.cc @@ -1,6 +1,7 @@ #include "daemon.hh" #include "monitor-fd.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "build-result.hh" #include "store-api.hh" #include "store-cast.hh" diff --git a/src/libstore/derivations.cc b/src/libstore/derivations.cc index ccb165d68..4a1f1e99f 100644 --- a/src/libstore/derivations.cc +++ b/src/libstore/derivations.cc @@ -5,6 +5,7 @@ #include "util.hh" #include "split.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "fs-accessor.hh" #include #include diff --git a/src/libstore/export-import.cc b/src/libstore/export-import.cc index 5ea263a86..72ad77124 100644 --- a/src/libstore/export-import.cc +++ b/src/libstore/export-import.cc @@ -2,6 +2,7 @@ #include "store-api.hh" #include "archive.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include diff --git a/src/libstore/legacy-ssh-store.cc b/src/libstore/legacy-ssh-store.cc index 2b7bebe9d..bff024f96 100644 --- a/src/libstore/legacy-ssh-store.cc +++ b/src/libstore/legacy-ssh-store.cc @@ -7,6 +7,7 @@ #include "store-api.hh" #include "path-with-outputs.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "ssh.hh" #include "derivations.hh" #include "callback.hh" diff --git a/src/libstore/path-info.cc b/src/libstore/path-info.cc index 97b72faa3..6b93976fa 100644 --- a/src/libstore/path-info.cc +++ b/src/libstore/path-info.cc @@ -1,5 +1,6 @@ #include "path-info.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "store-api.hh" namespace nix { diff --git a/src/libstore/remote-store.cc b/src/libstore/remote-store.cc index c3dfb5979..f02342d32 100644 --- a/src/libstore/remote-store.cc +++ b/src/libstore/remote-store.cc @@ -6,6 +6,7 @@ #include "build-result.hh" #include "remote-store.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "archive.hh" #include "globals.hh" #include "derivations.hh" diff --git a/src/libstore/worker-protocol-impl.hh b/src/libstore/worker-protocol-impl.hh new file mode 100644 index 000000000..509a8a5d0 --- /dev/null +++ b/src/libstore/worker-protocol-impl.hh @@ -0,0 +1,78 @@ +#pragma once +/** + * @file + * + * Template implementations (as opposed to mere declarations). + * + * This file is an exmample of the "impl.hh" pattern. See the + * contributing guide. + */ + +#include "worker-protocol.hh" + +namespace nix { + +template +std::vector WorkerProto>::read(const Store & store, Source & from) +{ + std::vector resSet; + auto size = readNum(from); + while (size--) { + resSet.push_back(WorkerProto::read(store, from)); + } + return resSet; +} + +template +void WorkerProto>::write(const Store & store, Sink & out, const std::vector & resSet) +{ + out << resSet.size(); + for (auto & key : resSet) { + WorkerProto::write(store, out, key); + } +} + +template +std::set WorkerProto>::read(const Store & store, Source & from) +{ + std::set resSet; + auto size = readNum(from); + while (size--) { + resSet.insert(WorkerProto::read(store, from)); + } + return resSet; +} + +template +void WorkerProto>::write(const Store & store, Sink & out, const std::set & resSet) +{ + out << resSet.size(); + for (auto & key : resSet) { + WorkerProto::write(store, out, key); + } +} + +template +std::map WorkerProto>::read(const Store & store, Source & from) +{ + std::map resMap; + auto size = readNum(from); + while (size--) { + auto k = WorkerProto::read(store, from); + auto v = WorkerProto::read(store, from); + resMap.insert_or_assign(std::move(k), std::move(v)); + } + return resMap; +} + +template +void WorkerProto>::write(const Store & store, Sink & out, const std::map & resMap) +{ + out << resMap.size(); + for (auto & i : resMap) { + WorkerProto::write(store, out, i.first); + WorkerProto::write(store, out, i.second); + } +} + +} diff --git a/src/libstore/worker-protocol.cc b/src/libstore/worker-protocol.cc index 51bb12026..49708b642 100644 --- a/src/libstore/worker-protocol.cc +++ b/src/libstore/worker-protocol.cc @@ -4,6 +4,7 @@ #include "store-api.hh" #include "build-result.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "archive.hh" #include "derivations.hh" diff --git a/src/libstore/worker-protocol.hh b/src/libstore/worker-protocol.hh index f06332d17..18a4e11b2 100644 --- a/src/libstore/worker-protocol.hh +++ b/src/libstore/worker-protocol.hh @@ -173,67 +173,4 @@ MAKE_WORKER_PROTO(std::optional); template<> MAKE_WORKER_PROTO(std::optional); -template -std::vector WorkerProto>::read(const Store & store, Source & from) -{ - std::vector resSet; - auto size = readNum(from); - while (size--) { - resSet.push_back(WorkerProto::read(store, from)); - } - return resSet; -} - -template -void WorkerProto>::write(const Store & store, Sink & out, const std::vector & resSet) -{ - out << resSet.size(); - for (auto & key : resSet) { - WorkerProto::write(store, out, key); - } -} - -template -std::set WorkerProto>::read(const Store & store, Source & from) -{ - std::set resSet; - auto size = readNum(from); - while (size--) { - resSet.insert(WorkerProto::read(store, from)); - } - return resSet; -} - -template -void WorkerProto>::write(const Store & store, Sink & out, const std::set & resSet) -{ - out << resSet.size(); - for (auto & key : resSet) { - WorkerProto::write(store, out, key); - } -} - -template -std::map WorkerProto>::read(const Store & store, Source & from) -{ - std::map resMap; - auto size = readNum(from); - while (size--) { - auto k = WorkerProto::read(store, from); - auto v = WorkerProto::read(store, from); - resMap.insert_or_assign(std::move(k), std::move(v)); - } - return resMap; -} - -template -void WorkerProto>::write(const Store & store, Sink & out, const std::map & resMap) -{ - out << resMap.size(); - for (auto & i : resMap) { - WorkerProto::write(store, out, i.first); - WorkerProto::write(store, out, i.second); - } -} - } diff --git a/src/libutil/config-impl.hh b/src/libutil/config-impl.hh index b6cae5ec3..d9726a907 100644 --- a/src/libutil/config-impl.hh +++ b/src/libutil/config-impl.hh @@ -4,6 +4,9 @@ * * Template implementations (as opposed to mere declarations). * + * This file is an exmample of the "impl.hh" pattern. See the + * contributing guide. + * * One only needs to include this when one is declaring a * `BaseClass` setting, or as derived class of such an * instantiation. diff --git a/src/nix-store/nix-store.cc b/src/nix-store/nix-store.cc index 61c189efb..94368e415 100644 --- a/src/nix-store/nix-store.cc +++ b/src/nix-store/nix-store.cc @@ -12,6 +12,7 @@ #include "shared.hh" #include "util.hh" #include "worker-protocol.hh" +#include "worker-protocol-impl.hh" #include "graphml.hh" #include "legacy.hh" #include "path-with-outputs.hh" From 95eae0c002da5ef35e583b46a67818ebb95d3a1e Mon Sep 17 00:00:00 2001 From: John Ericson Date: Fri, 26 May 2023 11:07:25 -0400 Subject: [PATCH 2/5] Put worker protocol items inside a `WorkerProto` struct 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 --- src/build-remote/build-remote.cc | 2 +- src/libstore/build/derivation-goal.cc | 4 +- src/libstore/daemon.cc | 148 ++++++++++---------- src/libstore/derivations.cc | 4 +- src/libstore/export-import.cc | 4 +- src/libstore/legacy-ssh-store.cc | 16 +-- src/libstore/path-info.cc | 4 +- src/libstore/remote-store.cc | 130 +++++++++--------- src/libstore/worker-protocol-impl.hh | 28 ++-- src/libstore/worker-protocol.cc | 56 ++++---- src/libstore/worker-protocol.hh | 186 ++++++++++++++++---------- src/nix-store/nix-store.cc | 18 +-- 12 files changed, 327 insertions(+), 273 deletions(-) diff --git a/src/build-remote/build-remote.cc b/src/build-remote/build-remote.cc index 323e04fdb..2fb17d06f 100644 --- a/src/build-remote/build-remote.cc +++ b/src/build-remote/build-remote.cc @@ -299,7 +299,7 @@ connected: !trusted || *trusted; }); - // See the very large comment in `case wopBuildDerivation:` in + // See the very large comment in `case WorkerProto::Op::BuildDerivation:` in // `src/libstore/daemon.cc` that explains the trust model here. // // This condition mirrors that: that code enforces the "rules" outlined there; diff --git a/src/libstore/build/derivation-goal.cc b/src/libstore/build/derivation-goal.cc index e02000c8b..1a946e3d2 100644 --- a/src/libstore/build/derivation-goal.cc +++ b/src/libstore/build/derivation-goal.cc @@ -1153,7 +1153,7 @@ HookReply DerivationGoal::tryBuildHook() /* Tell the hook all the inputs that have to be copied to the remote system. */ - workerProtoWrite(worker.store, hook->sink, inputPaths); + WorkerProto::write(worker.store, hook->sink, inputPaths); /* Tell the hooks the missing outputs that have to be copied back from the remote system. */ @@ -1164,7 +1164,7 @@ HookReply DerivationGoal::tryBuildHook() if (buildMode != bmCheck && status.known && status.known->isValid()) continue; missingOutputs.insert(outputName); } - workerProtoWrite(worker.store, hook->sink, missingOutputs); + WorkerProto::write(worker.store, hook->sink, missingOutputs); } hook->sink = FdSink(); diff --git a/src/libstore/daemon.cc b/src/libstore/daemon.cc index 54b089b30..7eba1a79d 100644 --- a/src/libstore/daemon.cc +++ b/src/libstore/daemon.cc @@ -264,7 +264,7 @@ static std::vector readDerivedPaths(Store & store, unsigned int cli { std::vector reqs; if (GET_PROTOCOL_MINOR(clientVersion) >= 30) { - reqs = WorkerProto>::read(store, from); + reqs = WorkerProto::Serialise>::read(store, from); } else { for (auto & s : readStrings(from)) reqs.push_back(parsePathWithOutputs(store, s).toDerivedPath()); @@ -274,11 +274,11 @@ static std::vector readDerivedPaths(Store & store, unsigned int cli static void performOp(TunnelLogger * logger, ref store, TrustedFlag trusted, RecursiveFlag recursive, unsigned int clientVersion, - Source & from, BufferedSink & to, unsigned int op) + Source & from, BufferedSink & to, WorkerProto::Op op) { switch (op) { - case wopIsValidPath: { + case WorkerProto::Op::IsValidPath: { auto path = store->parseStorePath(readString(from)); logger->startWork(); bool result = store->isValidPath(path); @@ -287,8 +287,8 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopQueryValidPaths: { - auto paths = WorkerProto::read(*store, from); + case WorkerProto::Op::QueryValidPaths: { + auto paths = WorkerProto::Serialise::read(*store, from); SubstituteFlag substitute = NoSubstitute; if (GET_PROTOCOL_MINOR(clientVersion) >= 27) { @@ -301,11 +301,11 @@ static void performOp(TunnelLogger * logger, ref store, } auto res = store->queryValidPaths(paths, substitute); logger->stopWork(); - workerProtoWrite(*store, to, res); + WorkerProto::write(*store, to, res); break; } - case wopHasSubstitutes: { + case WorkerProto::Op::HasSubstitutes: { auto path = store->parseStorePath(readString(from)); logger->startWork(); StorePathSet paths; // FIXME @@ -316,16 +316,16 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopQuerySubstitutablePaths: { - auto paths = WorkerProto::read(*store, from); + case WorkerProto::Op::QuerySubstitutablePaths: { + auto paths = WorkerProto::Serialise::read(*store, from); logger->startWork(); auto res = store->querySubstitutablePaths(paths); logger->stopWork(); - workerProtoWrite(*store, to, res); + WorkerProto::write(*store, to, res); break; } - case wopQueryPathHash: { + case WorkerProto::Op::QueryPathHash: { auto path = store->parseStorePath(readString(from)); logger->startWork(); auto hash = store->queryPathInfo(path)->narHash; @@ -334,27 +334,27 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopQueryReferences: - case wopQueryReferrers: - case wopQueryValidDerivers: - case wopQueryDerivationOutputs: { + case WorkerProto::Op::QueryReferences: + case WorkerProto::Op::QueryReferrers: + case WorkerProto::Op::QueryValidDerivers: + case WorkerProto::Op::QueryDerivationOutputs: { auto path = store->parseStorePath(readString(from)); logger->startWork(); StorePathSet paths; - if (op == wopQueryReferences) + if (op == WorkerProto::Op::QueryReferences) for (auto & i : store->queryPathInfo(path)->references) paths.insert(i); - else if (op == wopQueryReferrers) + else if (op == WorkerProto::Op::QueryReferrers) store->queryReferrers(path, paths); - else if (op == wopQueryValidDerivers) + else if (op == WorkerProto::Op::QueryValidDerivers) paths = store->queryValidDerivers(path); else paths = store->queryDerivationOutputs(path); logger->stopWork(); - workerProtoWrite(*store, to, paths); + WorkerProto::write(*store, to, paths); break; } - case wopQueryDerivationOutputNames: { + case WorkerProto::Op::QueryDerivationOutputNames: { auto path = store->parseStorePath(readString(from)); logger->startWork(); auto names = store->readDerivation(path).outputNames(); @@ -363,16 +363,16 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopQueryDerivationOutputMap: { + case WorkerProto::Op::QueryDerivationOutputMap: { auto path = store->parseStorePath(readString(from)); logger->startWork(); auto outputs = store->queryPartialDerivationOutputMap(path); logger->stopWork(); - workerProtoWrite(*store, to, outputs); + WorkerProto::write(*store, to, outputs); break; } - case wopQueryDeriver: { + case WorkerProto::Op::QueryDeriver: { auto path = store->parseStorePath(readString(from)); logger->startWork(); auto info = store->queryPathInfo(path); @@ -381,7 +381,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopQueryPathFromHashPart: { + case WorkerProto::Op::QueryPathFromHashPart: { auto hashPart = readString(from); logger->startWork(); auto path = store->queryPathFromHashPart(hashPart); @@ -390,11 +390,11 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopAddToStore: { + case WorkerProto::Op::AddToStore: { if (GET_PROTOCOL_MINOR(clientVersion) >= 25) { auto name = readString(from); auto camStr = readString(from); - auto refs = WorkerProto::read(*store, from); + auto refs = WorkerProto::Serialise::read(*store, from); bool repairBool; from >> repairBool; auto repair = RepairFlag{repairBool}; @@ -476,7 +476,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopAddMultipleToStore: { + case WorkerProto::Op::AddMultipleToStore: { bool repair, dontCheckSigs; from >> repair >> dontCheckSigs; if (!trusted && dontCheckSigs) @@ -493,10 +493,10 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopAddTextToStore: { + case WorkerProto::Op::AddTextToStore: { std::string suffix = readString(from); std::string s = readString(from); - auto refs = WorkerProto::read(*store, from); + auto refs = WorkerProto::Serialise::read(*store, from); logger->startWork(); auto path = store->addTextToStore(suffix, s, refs, NoRepair); logger->stopWork(); @@ -504,7 +504,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopExportPath: { + case WorkerProto::Op::ExportPath: { auto path = store->parseStorePath(readString(from)); readInt(from); // obsolete logger->startWork(); @@ -515,7 +515,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopImportPaths: { + case WorkerProto::Op::ImportPaths: { logger->startWork(); TunnelSource source(from, to); auto paths = store->importPaths(source, @@ -527,7 +527,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopBuildPaths: { + case WorkerProto::Op::BuildPaths: { auto drvs = readDerivedPaths(*store, clientVersion, from); BuildMode mode = bmNormal; if (GET_PROTOCOL_MINOR(clientVersion) >= 15) { @@ -552,7 +552,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopBuildPathsWithResults: { + case WorkerProto::Op::BuildPathsWithResults: { auto drvs = readDerivedPaths(*store, clientVersion, from); BuildMode mode = bmNormal; mode = (BuildMode) readInt(from); @@ -568,12 +568,12 @@ static void performOp(TunnelLogger * logger, ref store, auto results = store->buildPathsWithResults(drvs, mode); logger->stopWork(); - workerProtoWrite(*store, to, results); + WorkerProto::write(*store, to, results); break; } - case wopBuildDerivation: { + case WorkerProto::Op::BuildDerivation: { auto drvPath = store->parseStorePath(readString(from)); BasicDerivation drv; readDerivation(from, *store, drv, Derivation::nameFromPath(drvPath)); @@ -645,12 +645,12 @@ static void performOp(TunnelLogger * logger, ref store, DrvOutputs builtOutputs; for (auto & [output, realisation] : res.builtOutputs) builtOutputs.insert_or_assign(realisation.id, realisation); - workerProtoWrite(*store, to, builtOutputs); + WorkerProto::write(*store, to, builtOutputs); } break; } - case wopEnsurePath: { + case WorkerProto::Op::EnsurePath: { auto path = store->parseStorePath(readString(from)); logger->startWork(); store->ensurePath(path); @@ -659,7 +659,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopAddTempRoot: { + case WorkerProto::Op::AddTempRoot: { auto path = store->parseStorePath(readString(from)); logger->startWork(); store->addTempRoot(path); @@ -668,7 +668,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopAddIndirectRoot: { + case WorkerProto::Op::AddIndirectRoot: { Path path = absPath(readString(from)); logger->startWork(); @@ -681,14 +681,14 @@ static void performOp(TunnelLogger * logger, ref store, } // Obsolete. - case wopSyncWithGC: { + case WorkerProto::Op::SyncWithGC: { logger->startWork(); logger->stopWork(); to << 1; break; } - case wopFindRoots: { + case WorkerProto::Op::FindRoots: { logger->startWork(); auto & gcStore = require(*store); Roots roots = gcStore.findRoots(!trusted); @@ -707,10 +707,10 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopCollectGarbage: { + case WorkerProto::Op::CollectGarbage: { GCOptions options; options.action = (GCOptions::GCAction) readInt(from); - options.pathsToDelete = WorkerProto::read(*store, from); + options.pathsToDelete = WorkerProto::Serialise::read(*store, from); from >> options.ignoreLiveness >> options.maxFreed; // obsolete fields readInt(from); @@ -731,7 +731,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopSetOptions: { + case WorkerProto::Op::SetOptions: { ClientSettings clientSettings; @@ -768,7 +768,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopQuerySubstitutablePathInfo: { + case WorkerProto::Op::QuerySubstitutablePathInfo: { auto path = store->parseStorePath(readString(from)); logger->startWork(); SubstitutablePathInfos infos; @@ -780,22 +780,22 @@ static void performOp(TunnelLogger * logger, ref store, else { to << 1 << (i->second.deriver ? store->printStorePath(*i->second.deriver) : ""); - workerProtoWrite(*store, to, i->second.references); + WorkerProto::write(*store, to, i->second.references); to << i->second.downloadSize << i->second.narSize; } break; } - case wopQuerySubstitutablePathInfos: { + case WorkerProto::Op::QuerySubstitutablePathInfos: { SubstitutablePathInfos infos; StorePathCAMap pathsMap = {}; if (GET_PROTOCOL_MINOR(clientVersion) < 22) { - auto paths = WorkerProto::read(*store, from); + auto paths = WorkerProto::Serialise::read(*store, from); for (auto & path : paths) pathsMap.emplace(path, std::nullopt); } else - pathsMap = WorkerProto::read(*store, from); + pathsMap = WorkerProto::Serialise::read(*store, from); logger->startWork(); store->querySubstitutablePathInfos(pathsMap, infos); logger->stopWork(); @@ -803,21 +803,21 @@ static void performOp(TunnelLogger * logger, ref store, for (auto & i : infos) { to << store->printStorePath(i.first) << (i.second.deriver ? store->printStorePath(*i.second.deriver) : ""); - workerProtoWrite(*store, to, i.second.references); + WorkerProto::write(*store, to, i.second.references); to << i.second.downloadSize << i.second.narSize; } break; } - case wopQueryAllValidPaths: { + case WorkerProto::Op::QueryAllValidPaths: { logger->startWork(); auto paths = store->queryAllValidPaths(); logger->stopWork(); - workerProtoWrite(*store, to, paths); + WorkerProto::write(*store, to, paths); break; } - case wopQueryPathInfo: { + case WorkerProto::Op::QueryPathInfo: { auto path = store->parseStorePath(readString(from)); std::shared_ptr info; logger->startWork(); @@ -838,14 +838,14 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopOptimiseStore: + case WorkerProto::Op::OptimiseStore: logger->startWork(); store->optimiseStore(); logger->stopWork(); to << 1; break; - case wopVerifyStore: { + case WorkerProto::Op::VerifyStore: { bool checkContents, repair; from >> checkContents >> repair; logger->startWork(); @@ -857,7 +857,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopAddSignatures: { + case WorkerProto::Op::AddSignatures: { auto path = store->parseStorePath(readString(from)); StringSet sigs = readStrings(from); logger->startWork(); @@ -869,7 +869,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopNarFromPath: { + case WorkerProto::Op::NarFromPath: { auto path = store->parseStorePath(readString(from)); logger->startWork(); logger->stopWork(); @@ -877,7 +877,7 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopAddToStoreNar: { + case WorkerProto::Op::AddToStoreNar: { bool repair, dontCheckSigs; auto path = store->parseStorePath(readString(from)); auto deriver = readString(from); @@ -885,7 +885,7 @@ static void performOp(TunnelLogger * logger, ref store, ValidPathInfo info { path, narHash }; if (deriver != "") info.deriver = store->parseStorePath(deriver); - info.references = WorkerProto::read(*store, from); + info.references = WorkerProto::Serialise::read(*store, from); from >> info.registrationTime >> info.narSize >> info.ultimate; info.sigs = readStrings(from); info.ca = ContentAddress::parseOpt(readString(from)); @@ -929,21 +929,21 @@ static void performOp(TunnelLogger * logger, ref store, break; } - case wopQueryMissing: { + case WorkerProto::Op::QueryMissing: { auto targets = readDerivedPaths(*store, clientVersion, from); logger->startWork(); StorePathSet willBuild, willSubstitute, unknown; uint64_t downloadSize, narSize; store->queryMissing(targets, willBuild, willSubstitute, unknown, downloadSize, narSize); logger->stopWork(); - workerProtoWrite(*store, to, willBuild); - workerProtoWrite(*store, to, willSubstitute); - workerProtoWrite(*store, to, unknown); + WorkerProto::write(*store, to, willBuild); + WorkerProto::write(*store, to, willSubstitute); + WorkerProto::write(*store, to, unknown); to << downloadSize << narSize; break; } - case wopRegisterDrvOutput: { + case WorkerProto::Op::RegisterDrvOutput: { logger->startWork(); if (GET_PROTOCOL_MINOR(clientVersion) < 31) { auto outputId = DrvOutput::parse(readString(from)); @@ -951,14 +951,14 @@ static void performOp(TunnelLogger * logger, ref store, store->registerDrvOutput(Realisation{ .id = outputId, .outPath = outputPath}); } else { - auto realisation = WorkerProto::read(*store, from); + auto realisation = WorkerProto::Serialise::read(*store, from); store->registerDrvOutput(realisation); } logger->stopWork(); break; } - case wopQueryRealisation: { + case WorkerProto::Op::QueryRealisation: { logger->startWork(); auto outputId = DrvOutput::parse(readString(from)); auto info = store->queryRealisation(outputId); @@ -966,16 +966,16 @@ static void performOp(TunnelLogger * logger, ref store, if (GET_PROTOCOL_MINOR(clientVersion) < 31) { std::set outPaths; if (info) outPaths.insert(info->outPath); - workerProtoWrite(*store, to, outPaths); + WorkerProto::write(*store, to, outPaths); } else { std::set realisations; if (info) realisations.insert(*info); - workerProtoWrite(*store, to, realisations); + WorkerProto::write(*store, to, realisations); } break; } - case wopAddBuildLog: { + case WorkerProto::Op::AddBuildLog: { StorePath path{readString(from)}; logger->startWork(); if (!trusted) @@ -992,6 +992,10 @@ static void performOp(TunnelLogger * logger, ref store, break; } + case WorkerProto::Op::QueryFailedPaths: + case WorkerProto::Op::ClearFailedPaths: + throw Error("Removed operation %1%", op); + default: throw Error("invalid operation %1%", op); } @@ -1046,7 +1050,7 @@ void processConnection( auto temp = trusted ? store->isTrustedClient() : std::optional { NotTrusted }; - workerProtoWrite(*store, to, temp); + WorkerProto::write(*store, to, temp); } /* Send startup error messages to the client. */ @@ -1059,9 +1063,9 @@ void processConnection( /* Process client requests. */ while (true) { - WorkerOp op; + WorkerProto::Op op; try { - op = (WorkerOp) readInt(from); + op = (enum WorkerProto::Op) readInt(from); } catch (Interrupted & e) { break; } catch (EndOfFile & e) { diff --git a/src/libstore/derivations.cc b/src/libstore/derivations.cc index 4a1f1e99f..2295ff9e8 100644 --- a/src/libstore/derivations.cc +++ b/src/libstore/derivations.cc @@ -750,7 +750,7 @@ Source & readDerivation(Source & in, const Store & store, BasicDerivation & drv, drv.outputs.emplace(std::move(name), std::move(output)); } - drv.inputSrcs = WorkerProto::read(store, in); + drv.inputSrcs = WorkerProto::Serialise::read(store, in); in >> drv.platform >> drv.builder; drv.args = readStrings(in); @@ -798,7 +798,7 @@ void writeDerivation(Sink & out, const Store & store, const BasicDerivation & dr }, }, i.second.raw()); } - workerProtoWrite(store, out, drv.inputSrcs); + WorkerProto::write(store, out, drv.inputSrcs); out << drv.platform << drv.builder << drv.args; out << drv.env.size(); for (auto & i : drv.env) diff --git a/src/libstore/export-import.cc b/src/libstore/export-import.cc index 72ad77124..b4100bace 100644 --- a/src/libstore/export-import.cc +++ b/src/libstore/export-import.cc @@ -46,7 +46,7 @@ void Store::exportPath(const StorePath & path, Sink & sink) teeSink << exportMagic << printStorePath(path); - workerProtoWrite(*this, teeSink, info->references); + WorkerProto::write(*this, teeSink, info->references); teeSink << (info->deriver ? printStorePath(*info->deriver) : "") << 0; @@ -74,7 +74,7 @@ StorePaths Store::importPaths(Source & source, CheckSigsFlag checkSigs) //Activity act(*logger, lvlInfo, "importing path '%s'", info.path); - auto references = WorkerProto::read(*this, source); + auto references = WorkerProto::Serialise::read(*this, source); auto deriver = readString(source); auto narHash = hashString(htSHA256, saved.s); diff --git a/src/libstore/legacy-ssh-store.cc b/src/libstore/legacy-ssh-store.cc index bff024f96..7d006f451 100644 --- a/src/libstore/legacy-ssh-store.cc +++ b/src/libstore/legacy-ssh-store.cc @@ -147,7 +147,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor auto deriver = readString(conn->from); if (deriver != "") info->deriver = parseStorePath(deriver); - info->references = WorkerProto::read(*this, conn->from); + info->references = WorkerProto::Serialise::read(*this, conn->from); readLongLong(conn->from); // download size info->narSize = readLongLong(conn->from); @@ -181,7 +181,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor << 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 @@ -210,7 +210,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor conn->to << exportMagic << printStorePath(info.path); - workerProtoWrite(*this, conn->to, info.references); + WorkerProto::write(*this, conn->to, info.references); conn->to << (info.deriver ? printStorePath(*info.deriver) : "") << 0 @@ -295,7 +295,7 @@ public: if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 3) conn->from >> status.timesBuilt >> status.isNonDeterministic >> status.startTime >> status.stopTime; if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 6) { - auto builtOutputs = WorkerProto::read(*this, conn->from); + auto builtOutputs = WorkerProto::Serialise::read(*this, conn->from); for (auto && [output, realisation] : builtOutputs) status.builtOutputs.insert_or_assign( std::move(output.outputName), @@ -370,10 +370,10 @@ public: conn->to << cmdQueryClosure << includeOutputs; - workerProtoWrite(*this, conn->to, paths); + WorkerProto::write(*this, conn->to, paths); conn->to.flush(); - for (auto & i : WorkerProto::read(*this, conn->from)) + for (auto & i : WorkerProto::Serialise::read(*this, conn->from)) out.insert(i); } @@ -386,10 +386,10 @@ public: << cmdQueryValidPaths << false // lock << maybeSubstitute; - workerProtoWrite(*this, conn->to, paths); + WorkerProto::write(*this, conn->to, paths); conn->to.flush(); - return WorkerProto::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, conn->from); } void connect() override diff --git a/src/libstore/path-info.cc b/src/libstore/path-info.cc index 6b93976fa..9971de1dd 100644 --- a/src/libstore/path-info.cc +++ b/src/libstore/path-info.cc @@ -133,7 +133,7 @@ ValidPathInfo ValidPathInfo::read(Source & source, const Store & store, unsigned auto narHash = Hash::parseAny(readString(source), htSHA256); ValidPathInfo info(path, narHash); if (deriver != "") info.deriver = store.parseStorePath(deriver); - info.references = WorkerProto::read(store, source); + info.references = WorkerProto::Serialise::read(store, source); source >> info.registrationTime >> info.narSize; if (format >= 16) { source >> info.ultimate; @@ -154,7 +154,7 @@ void ValidPathInfo::write( sink << store.printStorePath(path); sink << (deriver ? store.printStorePath(*deriver) : "") << narHash.to_string(Base16, false); - workerProtoWrite(store, sink, references); + WorkerProto::write(store, sink, references); sink << registrationTime << narSize; if (format >= 16) { sink << ultimate diff --git a/src/libstore/remote-store.cc b/src/libstore/remote-store.cc index f02342d32..9253ce09e 100644 --- a/src/libstore/remote-store.cc +++ b/src/libstore/remote-store.cc @@ -101,7 +101,7 @@ void RemoteStore::initConnection(Connection & conn) } if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 35) { - conn.remoteTrustsUs = WorkerProto>::read(*this, conn.from); + conn.remoteTrustsUs = WorkerProto::Serialise>::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::read(*this, conn->from); + return WorkerProto::Serialise::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::read(*this, conn->from); + return WorkerProto::Serialise::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::read(*this, conn->from); + return WorkerProto::Serialise::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::read(*this, conn->from); + info.references = WorkerProto::Serialise::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(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::read(*this, conn->from); + info.references = WorkerProto::Serialise::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 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::read(*this, conn->from)) + for (auto & i : WorkerProto::Serialise::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::read(*this, conn->from); + return WorkerProto::Serialise::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::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, conn->from); } @@ -380,9 +380,9 @@ std::map> 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>>::read(*this, conn->from); + return WorkerProto::Serialise>>::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> RemoteStore::queryPartialDerivat std::optional 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 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 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::read(*this, conn->from); + auto importedPaths = WorkerProto::Serialise::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 { if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) { - auto outPaths = WorkerProto>::read( + auto outPaths = WorkerProto::Serialise>::read( *this, conn->from); if (outPaths.empty()) return nullptr; return std::make_shared(Realisation { .id = id, .outPath = *outPaths.begin() }); } else { - auto realisations = WorkerProto>::read( + auto realisations = WorkerProto::Serialise>::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 & 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 & 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 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>::read(*this, conn->from); + return WorkerProto::Serialise>::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::read(*this, conn->from); + auto builtOutputs = WorkerProto::Serialise::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(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 & 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::read(*this, conn->from); - willSubstitute = WorkerProto::read(*this, conn->from); - unknown = WorkerProto::read(*this, conn->from); + willBuild = WorkerProto::Serialise::read(*this, conn->from); + willSubstitute = WorkerProto::Serialise::read(*this, conn->from); + unknown = WorkerProto::Serialise::read(*this, conn->from); conn->from >> downloadSize >> narSize; return; } @@ -941,7 +941,7 @@ void RemoteStore::queryMissing(const std::vector & 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); } diff --git a/src/libstore/worker-protocol-impl.hh b/src/libstore/worker-protocol-impl.hh index 509a8a5d0..f34e55c54 100644 --- a/src/libstore/worker-protocol-impl.hh +++ b/src/libstore/worker-protocol-impl.hh @@ -13,65 +13,65 @@ namespace nix { template -std::vector WorkerProto>::read(const Store & store, Source & from) +std::vector WorkerProto::Serialise>::read(const Store & store, Source & from) { std::vector resSet; auto size = readNum(from); while (size--) { - resSet.push_back(WorkerProto::read(store, from)); + resSet.push_back(WorkerProto::Serialise::read(store, from)); } return resSet; } template -void WorkerProto>::write(const Store & store, Sink & out, const std::vector & resSet) +void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::vector & resSet) { out << resSet.size(); for (auto & key : resSet) { - WorkerProto::write(store, out, key); + WorkerProto::Serialise::write(store, out, key); } } template -std::set WorkerProto>::read(const Store & store, Source & from) +std::set WorkerProto::Serialise>::read(const Store & store, Source & from) { std::set resSet; auto size = readNum(from); while (size--) { - resSet.insert(WorkerProto::read(store, from)); + resSet.insert(WorkerProto::Serialise::read(store, from)); } return resSet; } template -void WorkerProto>::write(const Store & store, Sink & out, const std::set & resSet) +void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::set & resSet) { out << resSet.size(); for (auto & key : resSet) { - WorkerProto::write(store, out, key); + WorkerProto::Serialise::write(store, out, key); } } template -std::map WorkerProto>::read(const Store & store, Source & from) +std::map WorkerProto::Serialise>::read(const Store & store, Source & from) { std::map resMap; auto size = readNum(from); while (size--) { - auto k = WorkerProto::read(store, from); - auto v = WorkerProto::read(store, from); + auto k = WorkerProto::Serialise::read(store, from); + auto v = WorkerProto::Serialise::read(store, from); resMap.insert_or_assign(std::move(k), std::move(v)); } return resMap; } template -void WorkerProto>::write(const Store & store, Sink & out, const std::map & resMap) +void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::map & resMap) { out << resMap.size(); for (auto & i : resMap) { - WorkerProto::write(store, out, i.first); - WorkerProto::write(store, out, i.second); + WorkerProto::Serialise::write(store, out, i.first); + WorkerProto::Serialise::write(store, out, i.second); } } diff --git a/src/libstore/worker-protocol.cc b/src/libstore/worker-protocol.cc index 49708b642..0bf9e4d68 100644 --- a/src/libstore/worker-protocol.cc +++ b/src/libstore/worker-protocol.cc @@ -12,29 +12,29 @@ namespace nix { -std::string WorkerProto::read(const Store & store, Source & from) +std::string WorkerProto::Serialise::read(const Store & store, Source & from) { return readString(from); } -void WorkerProto::write(const Store & store, Sink & out, const std::string & str) +void WorkerProto::Serialise::write(const Store & store, Sink & out, const std::string & str) { out << str; } -StorePath WorkerProto::read(const Store & store, Source & from) +StorePath WorkerProto::Serialise::read(const Store & store, Source & from) { return store.parseStorePath(readString(from)); } -void WorkerProto::write(const Store & store, Sink & out, const StorePath & storePath) +void WorkerProto::Serialise::write(const Store & store, Sink & out, const StorePath & storePath) { out << store.printStorePath(storePath); } -std::optional WorkerProto>::read(const Store & store, Source & from) +std::optional WorkerProto::Serialise>::read(const Store & store, Source & from) { auto temp = readNum(from); switch (temp) { @@ -49,7 +49,7 @@ std::optional WorkerProto>::read(const S } } -void WorkerProto>::write(const Store & store, Sink & out, const std::optional & optTrusted) +void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::optional & optTrusted) { if (!optTrusted) out << (uint8_t)0; @@ -68,30 +68,30 @@ void WorkerProto>::write(const Store & store, Sink & } -ContentAddress WorkerProto::read(const Store & store, Source & from) +ContentAddress WorkerProto::Serialise::read(const Store & store, Source & from) { return ContentAddress::parse(readString(from)); } -void WorkerProto::write(const Store & store, Sink & out, const ContentAddress & ca) +void WorkerProto::Serialise::write(const Store & store, Sink & out, const ContentAddress & ca) { out << renderContentAddress(ca); } -DerivedPath WorkerProto::read(const Store & store, Source & from) +DerivedPath WorkerProto::Serialise::read(const Store & store, Source & from) { auto s = readString(from); return DerivedPath::parseLegacy(store, s); } -void WorkerProto::write(const Store & store, Sink & out, const DerivedPath & req) +void WorkerProto::Serialise::write(const Store & store, Sink & out, const DerivedPath & req) { out << req.to_string_legacy(store); } -Realisation WorkerProto::read(const Store & store, Source & from) +Realisation WorkerProto::Serialise::read(const Store & store, Source & from) { std::string rawInput = readString(from); return Realisation::fromJSON( @@ -100,41 +100,41 @@ Realisation WorkerProto::read(const Store & store, Source & from) ); } -void WorkerProto::write(const Store & store, Sink & out, const Realisation & realisation) +void WorkerProto::Serialise::write(const Store & store, Sink & out, const Realisation & realisation) { out << realisation.toJSON().dump(); } -DrvOutput WorkerProto::read(const Store & store, Source & from) +DrvOutput WorkerProto::Serialise::read(const Store & store, Source & from) { return DrvOutput::parse(readString(from)); } -void WorkerProto::write(const Store & store, Sink & out, const DrvOutput & drvOutput) +void WorkerProto::Serialise::write(const Store & store, Sink & out, const DrvOutput & drvOutput) { out << drvOutput.to_string(); } -KeyedBuildResult WorkerProto::read(const Store & store, Source & from) +KeyedBuildResult WorkerProto::Serialise::read(const Store & store, Source & from) { - auto path = WorkerProto::read(store, from); - auto br = WorkerProto::read(store, from); + auto path = WorkerProto::Serialise::read(store, from); + auto br = WorkerProto::Serialise::read(store, from); return KeyedBuildResult { std::move(br), /* .path = */ std::move(path), }; } -void WorkerProto::write(const Store & store, Sink & to, const KeyedBuildResult & res) +void WorkerProto::Serialise::write(const Store & store, Sink & to, const KeyedBuildResult & res) { - workerProtoWrite(store, to, res.path); - workerProtoWrite(store, to, static_cast(res)); + WorkerProto::write(store, to, res.path); + WorkerProto::write(store, to, static_cast(res)); } -BuildResult WorkerProto::read(const Store & store, Source & from) +BuildResult WorkerProto::Serialise::read(const Store & store, Source & from) { BuildResult res; res.status = (BuildResult::Status) readInt(from); @@ -144,7 +144,7 @@ BuildResult WorkerProto::read(const Store & store, Source & from) >> res.isNonDeterministic >> res.startTime >> res.stopTime; - auto builtOutputs = WorkerProto::read(store, from); + auto builtOutputs = WorkerProto::Serialise::read(store, from); for (auto && [output, realisation] : builtOutputs) res.builtOutputs.insert_or_assign( std::move(output.outputName), @@ -152,7 +152,7 @@ BuildResult WorkerProto::read(const Store & store, Source & from) return res; } -void WorkerProto::write(const Store & store, Sink & to, const BuildResult & res) +void WorkerProto::Serialise::write(const Store & store, Sink & to, const BuildResult & res) { to << res.status @@ -164,28 +164,28 @@ void WorkerProto::write(const Store & store, Sink & to, const Build DrvOutputs builtOutputs; for (auto & [output, realisation] : res.builtOutputs) builtOutputs.insert_or_assign(realisation.id, realisation); - workerProtoWrite(store, to, builtOutputs); + WorkerProto::write(store, to, builtOutputs); } -std::optional WorkerProto>::read(const Store & store, Source & from) +std::optional WorkerProto::Serialise>::read(const Store & store, Source & from) { auto s = readString(from); return s == "" ? std::optional {} : store.parseStorePath(s); } -void WorkerProto>::write(const Store & store, Sink & out, const std::optional & storePathOpt) +void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::optional & storePathOpt) { out << (storePathOpt ? store.printStorePath(*storePathOpt) : ""); } -std::optional WorkerProto>::read(const Store & store, Source & from) +std::optional WorkerProto::Serialise>::read(const Store & store, Source & from) { return ContentAddress::parseOpt(readString(from)); } -void WorkerProto>::write(const Store & store, Sink & out, const std::optional & caOpt) +void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::optional & caOpt) { out << (caOpt ? renderContentAddress(*caOpt) : ""); } diff --git a/src/libstore/worker-protocol.hh b/src/libstore/worker-protocol.hh index 18a4e11b2..cd6801290 100644 --- a/src/libstore/worker-protocol.hh +++ b/src/libstore/worker-protocol.hh @@ -14,57 +14,6 @@ namespace nix { #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff) -/** - * Enumeration of all the request types for the "worker protocol", used - * by unix:// and ssh-ng:// stores. - */ -typedef enum { - wopIsValidPath = 1, - wopHasSubstitutes = 3, - wopQueryPathHash = 4, // obsolete - wopQueryReferences = 5, // obsolete - wopQueryReferrers = 6, - wopAddToStore = 7, - wopAddTextToStore = 8, // obsolete since 1.25, Nix 3.0. Use wopAddToStore - wopBuildPaths = 9, - wopEnsurePath = 10, - wopAddTempRoot = 11, - wopAddIndirectRoot = 12, - wopSyncWithGC = 13, - wopFindRoots = 14, - wopExportPath = 16, // obsolete - wopQueryDeriver = 18, // obsolete - wopSetOptions = 19, - wopCollectGarbage = 20, - wopQuerySubstitutablePathInfo = 21, - wopQueryDerivationOutputs = 22, // obsolete - wopQueryAllValidPaths = 23, - wopQueryFailedPaths = 24, - wopClearFailedPaths = 25, - wopQueryPathInfo = 26, - wopImportPaths = 27, // obsolete - wopQueryDerivationOutputNames = 28, // obsolete - wopQueryPathFromHashPart = 29, - wopQuerySubstitutablePathInfos = 30, - wopQueryValidPaths = 31, - wopQuerySubstitutablePaths = 32, - wopQueryValidDerivers = 33, - wopOptimiseStore = 34, - wopVerifyStore = 35, - wopBuildDerivation = 36, - wopAddSignatures = 37, - wopNarFromPath = 38, - wopAddToStoreNar = 39, - wopQueryMissing = 40, - wopQueryDerivationOutputMap = 41, - wopRegisterDrvOutput = 42, - wopQueryRealisation = 43, - wopAddMultipleToStore = 44, - wopAddBuildLog = 45, - wopBuildPathsWithResults = 46, -} WorkerOp; - - #define STDERR_NEXT 0x6f6c6d67 #define STDERR_READ 0x64617461 // data needed from source #define STDERR_WRITE 0x64617416 // data for sink @@ -78,7 +27,7 @@ typedef enum { class Store; struct Source; -// items being serialized +// items being serialised struct DerivedPath; struct DrvOutput; struct Realisation; @@ -88,31 +37,132 @@ enum TrustedFlag : bool; /** - * Data type for canonical pairs of serializers for the worker protocol. + * The "worker protocol", used by unix:// and ssh-ng:// stores. * - * See https://en.cppreference.com/w/cpp/language/adl for the broader - * concept of what is going on here. + * This `struct` is basically just a `namespace`; We use a type rather + * than a namespace just so we can use it as a template argument. */ -template -struct WorkerProto { - static T read(const Store & store, Source & from); - static void write(const Store & store, Sink & out, const T & t); +struct WorkerProto +{ + /** + * Enumeration of all the request types for the protocol. + */ + enum struct Op : uint64_t; + + /** + * Data type for canonical pairs of serialisers for the worker protocol. + * + * See https://en.cppreference.com/w/cpp/language/adl for the broader + * concept of what is going on here. + */ + template + struct Serialise; + // This is the definition of `Serialise` we *want* to put here, but + // do not do so. + // + // The problem is that if we do so, C++ will think we have + // seralisers for *all* types. We don't, of course, but that won't + // cause an error until link time. That makes for long debug cycles + // when there is a missing serialiser. + // + // By not defining it globally, and instead letting individual + // serialisers specialise the type, we get back the compile-time + // errors we would like. When no serialiser exists, C++ sees an + // abstract "incomplete" type with no definition, and any attempt to + // use `to` or `from` static methods is a compile-time error because + // they don't exist on an incomplete type. + // + // This makes for a quicker debug cycle, as desired. +#if 0 + { + static T read(const Store & store, Source & from); + static void write(const Store & store, Sink & out, const T & t); + }; +#endif + + /** + * Wrapper function around `WorkerProto::Serialise::write` that allows us to + * infer the type instead of having to write it down explicitly. + */ + template + static void write(const Store & store, Sink & out, const T & t) + { + WorkerProto::Serialise::write(store, out, t); + } +}; + +enum struct WorkerProto::Op : uint64_t +{ + IsValidPath = 1, + HasSubstitutes = 3, + QueryPathHash = 4, // obsolete + QueryReferences = 5, // obsolete + QueryReferrers = 6, + AddToStore = 7, + AddTextToStore = 8, // obsolete since 1.25, Nix 3.0. Use WorkerProto::Op::AddToStore + BuildPaths = 9, + EnsurePath = 10, + AddTempRoot = 11, + AddIndirectRoot = 12, + SyncWithGC = 13, + FindRoots = 14, + ExportPath = 16, // obsolete + QueryDeriver = 18, // obsolete + SetOptions = 19, + CollectGarbage = 20, + QuerySubstitutablePathInfo = 21, + QueryDerivationOutputs = 22, // obsolete + QueryAllValidPaths = 23, + QueryFailedPaths = 24, + ClearFailedPaths = 25, + QueryPathInfo = 26, + ImportPaths = 27, // obsolete + QueryDerivationOutputNames = 28, // obsolete + QueryPathFromHashPart = 29, + QuerySubstitutablePathInfos = 30, + QueryValidPaths = 31, + QuerySubstitutablePaths = 32, + QueryValidDerivers = 33, + OptimiseStore = 34, + VerifyStore = 35, + BuildDerivation = 36, + AddSignatures = 37, + NarFromPath = 38, + AddToStoreNar = 39, + QueryMissing = 40, + QueryDerivationOutputMap = 41, + RegisterDrvOutput = 42, + QueryRealisation = 43, + AddMultipleToStore = 44, + AddBuildLog = 45, + BuildPathsWithResults = 46, }; /** - * Wrapper function around `WorkerProto::write` that allows us to - * infer the type instead of having to write it down explicitly. + * Convenience for sending operation codes. + * + * @todo Switch to using `WorkerProto::Serialise` instead probably. But + * this was not done at this time so there would be less churn. */ -template -void workerProtoWrite(const Store & store, Sink & out, const T & t) +inline Sink & operator << (Sink & sink, WorkerProto::Op op) { - WorkerProto::write(store, out, t); + return sink << (uint64_t) op; } /** - * Declare a canonical serializer pair for the worker protocol. + * Convenience for debugging. * - * We specialize the struct merely to indicate that we are implementing + * @todo Perhaps render known opcodes more nicely. + */ +inline std::ostream & operator << (std::ostream & s, WorkerProto::Op op) +{ + return s << (uint64_t) op; +} + +/** + * Declare a canonical serialiser pair for the worker protocol. + * + * We specialise the struct merely to indicate that we are implementing * the function for the given type. * * Some sort of `template<...>` must be used with the caller for this to @@ -120,7 +170,7 @@ void workerProtoWrite(const Store & store, Sink & out, const T & t) * practice. */ #define MAKE_WORKER_PROTO(T) \ - struct WorkerProto< T > { \ + struct WorkerProto::Serialise< T > { \ static T read(const Store & store, Source & from); \ static void write(const Store & store, Sink & out, const T & t); \ }; @@ -156,7 +206,7 @@ MAKE_WORKER_PROTO(X_); /** * These use the empty string for the null case, relying on the fact - * that the underlying types never serialize to the empty string. + * that the underlying types never serialise to the empty string. * * We do this instead of a generic std::optional instance because * ordinal tags (0 or 1, here) are a bit of a compatability hazard. For diff --git a/src/nix-store/nix-store.cc b/src/nix-store/nix-store.cc index 94368e415..1558ac11e 100644 --- a/src/nix-store/nix-store.cc +++ b/src/nix-store/nix-store.cc @@ -850,7 +850,7 @@ static void opServe(Strings opFlags, Strings opArgs) case cmdQueryValidPaths: { bool lock = readInt(in); bool substitute = readInt(in); - auto paths = WorkerProto::read(*store, in); + auto paths = WorkerProto::Serialise::read(*store, in); if (lock && writeAllowed) for (auto & path : paths) store->addTempRoot(path); @@ -859,19 +859,19 @@ static void opServe(Strings opFlags, Strings opArgs) store->substitutePaths(paths); } - workerProtoWrite(*store, out, store->queryValidPaths(paths)); + WorkerProto::write(*store, out, store->queryValidPaths(paths)); break; } case cmdQueryPathInfos: { - auto paths = WorkerProto::read(*store, in); + auto paths = WorkerProto::Serialise::read(*store, in); // !!! Maybe we want a queryPathInfos? for (auto & i : paths) { try { auto info = store->queryPathInfo(i); out << store->printStorePath(info->path) << (info->deriver ? store->printStorePath(*info->deriver) : ""); - workerProtoWrite(*store, out, info->references); + WorkerProto::write(*store, out, info->references); // !!! Maybe we want compression? out << info->narSize // downloadSize << info->narSize; @@ -899,7 +899,7 @@ static void opServe(Strings opFlags, Strings opArgs) case cmdExportPaths: { readInt(in); // obsolete - store->exportPaths(WorkerProto::read(*store, in), out); + store->exportPaths(WorkerProto::Serialise::read(*store, in), out); break; } @@ -945,7 +945,7 @@ static void opServe(Strings opFlags, Strings opArgs) DrvOutputs builtOutputs; for (auto & [output, realisation] : status.builtOutputs) builtOutputs.insert_or_assign(realisation.id, realisation); - workerProtoWrite(*store, out, builtOutputs); + WorkerProto::write(*store, out, builtOutputs); } break; @@ -954,9 +954,9 @@ static void opServe(Strings opFlags, Strings opArgs) case cmdQueryClosure: { bool includeOutputs = readInt(in); StorePathSet closure; - store->computeFSClosure(WorkerProto::read(*store, in), + store->computeFSClosure(WorkerProto::Serialise::read(*store, in), closure, false, includeOutputs); - workerProtoWrite(*store, out, closure); + WorkerProto::write(*store, out, closure); break; } @@ -971,7 +971,7 @@ static void opServe(Strings opFlags, Strings opArgs) }; if (deriver != "") info.deriver = store->parseStorePath(deriver); - info.references = WorkerProto::read(*store, in); + info.references = WorkerProto::Serialise::read(*store, in); in >> info.registrationTime >> info.narSize >> info.ultimate; info.sigs = readStrings(in); info.ca = ContentAddress::parseOpt(readString(in)); From 4e8b495ad7dddabc35bf9d6afe3573426ffed15d Mon Sep 17 00:00:00 2001 From: John Ericson Date: Fri, 26 May 2023 11:22:24 -0400 Subject: [PATCH 3/5] Likewise namespace and `enum struct`-ify `ServeCommand` The motivation is exactly the same as for the last commit. In addition, this anticipates us formally defining separate serialisers for the serve protocol. --- src/libstore/legacy-ssh-store.cc | 16 ++++----- src/libstore/serve-protocol.hh | 58 ++++++++++++++++++++++++++------ src/nix-store/nix-store.cc | 22 ++++++------ 3 files changed, 66 insertions(+), 30 deletions(-) diff --git a/src/libstore/legacy-ssh-store.cc b/src/libstore/legacy-ssh-store.cc index 7d006f451..55ecab0ff 100644 --- a/src/libstore/legacy-ssh-store.cc +++ b/src/libstore/legacy-ssh-store.cc @@ -134,7 +134,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor debug("querying remote host '%s' for info on '%s'", host, printStorePath(path)); - conn->to << cmdQueryPathInfos << PathSet{printStorePath(path)}; + conn->to << ServeProto::Command::QueryPathInfos << PathSet{printStorePath(path)}; conn->to.flush(); auto p = readString(conn->from); @@ -177,7 +177,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 5) { conn->to - << cmdAddToStoreNar + << ServeProto::Command::AddToStoreNar << printStorePath(info.path) << (info.deriver ? printStorePath(*info.deriver) : "") << info.narHash.to_string(Base16, false); @@ -199,7 +199,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor } else { conn->to - << cmdImportPaths + << ServeProto::Command::ImportPaths << 1; try { copyNAR(source, conn->to); @@ -227,7 +227,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor { auto conn(connections->get()); - conn->to << cmdDumpStorePath << printStorePath(path); + conn->to << ServeProto::Command::DumpStorePath << printStorePath(path); conn->to.flush(); copyNAR(conn->from, sink); } @@ -280,7 +280,7 @@ public: auto conn(connections->get()); conn->to - << cmdBuildDerivation + << ServeProto::Command::BuildDerivation << printStorePath(drvPath); writeDerivation(conn->to, *this, drv); @@ -311,7 +311,7 @@ public: auto conn(connections->get()); - conn->to << cmdBuildPaths; + conn->to << ServeProto::Command::BuildPaths; Strings ss; for (auto & p : drvPaths) { auto sOrDrvPath = StorePathWithOutputs::tryFromDerivedPath(p); @@ -368,7 +368,7 @@ public: auto conn(connections->get()); conn->to - << cmdQueryClosure + << ServeProto::Command::QueryClosure << includeOutputs; WorkerProto::write(*this, conn->to, paths); conn->to.flush(); @@ -383,7 +383,7 @@ public: auto conn(connections->get()); conn->to - << cmdQueryValidPaths + << ServeProto::Command::QueryValidPaths << false // lock << maybeSubstitute; WorkerProto::write(*this, conn->to, paths); diff --git a/src/libstore/serve-protocol.hh b/src/libstore/serve-protocol.hh index 553fd3a09..7e43b3969 100644 --- a/src/libstore/serve-protocol.hh +++ b/src/libstore/serve-protocol.hh @@ -10,16 +10,52 @@ namespace nix { #define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00) #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff) -typedef enum { - cmdQueryValidPaths = 1, - cmdQueryPathInfos = 2, - cmdDumpStorePath = 3, - cmdImportPaths = 4, - cmdExportPaths = 5, - cmdBuildPaths = 6, - cmdQueryClosure = 7, - cmdBuildDerivation = 8, - cmdAddToStoreNar = 9, -} ServeCommand; +/** + * The "serve protocol", used by ssh:// stores. + * + * This `struct` is basically just a `namespace`; We use a type rather + * than a namespace just so we can use it as a template argument. + */ +struct ServeProto +{ + /** + * Enumeration of all the request types for the protocol. + */ + enum struct Command : uint64_t; +}; + +enum struct ServeProto::Command : uint64_t +{ + QueryValidPaths = 1, + QueryPathInfos = 2, + DumpStorePath = 3, + ImportPaths = 4, + ExportPaths = 5, + BuildPaths = 6, + QueryClosure = 7, + BuildDerivation = 8, + AddToStoreNar = 9, +}; + +/** + * Convenience for sending operation codes. + * + * @todo Switch to using `ServeProto::Serialize` instead probably. But + * this was not done at this time so there would be less churn. + */ +inline Sink & operator << (Sink & sink, ServeProto::Command op) +{ + return sink << (uint64_t) op; +} + +/** + * Convenience for debugging. + * + * @todo Perhaps render known opcodes more nicely. + */ +inline std::ostream & operator << (std::ostream & s, ServeProto::Command op) +{ + return s << (uint64_t) op; +} } diff --git a/src/nix-store/nix-store.cc b/src/nix-store/nix-store.cc index 1558ac11e..062c68ff8 100644 --- a/src/nix-store/nix-store.cc +++ b/src/nix-store/nix-store.cc @@ -838,16 +838,16 @@ static void opServe(Strings opFlags, Strings opArgs) }; while (true) { - ServeCommand cmd; + ServeProto::Command cmd; try { - cmd = (ServeCommand) readInt(in); + cmd = (ServeProto::Command) readInt(in); } catch (EndOfFile & e) { break; } switch (cmd) { - case cmdQueryValidPaths: { + case ServeProto::Command::QueryValidPaths: { bool lock = readInt(in); bool substitute = readInt(in); auto paths = WorkerProto::Serialise::read(*store, in); @@ -863,7 +863,7 @@ static void opServe(Strings opFlags, Strings opArgs) break; } - case cmdQueryPathInfos: { + case ServeProto::Command::QueryPathInfos: { auto paths = WorkerProto::Serialise::read(*store, in); // !!! Maybe we want a queryPathInfos? for (auto & i : paths) { @@ -886,24 +886,24 @@ static void opServe(Strings opFlags, Strings opArgs) break; } - case cmdDumpStorePath: + case ServeProto::Command::DumpStorePath: store->narFromPath(store->parseStorePath(readString(in)), out); break; - case cmdImportPaths: { + case ServeProto::Command::ImportPaths: { if (!writeAllowed) throw Error("importing paths is not allowed"); store->importPaths(in, NoCheckSigs); // FIXME: should we skip sig checking? out << 1; // indicate success break; } - case cmdExportPaths: { + case ServeProto::Command::ExportPaths: { readInt(in); // obsolete store->exportPaths(WorkerProto::Serialise::read(*store, in), out); break; } - case cmdBuildPaths: { + case ServeProto::Command::BuildPaths: { if (!writeAllowed) throw Error("building paths is not allowed"); @@ -924,7 +924,7 @@ static void opServe(Strings opFlags, Strings opArgs) break; } - case cmdBuildDerivation: { /* Used by hydra-queue-runner. */ + case ServeProto::Command::BuildDerivation: { /* Used by hydra-queue-runner. */ if (!writeAllowed) throw Error("building paths is not allowed"); @@ -951,7 +951,7 @@ static void opServe(Strings opFlags, Strings opArgs) break; } - case cmdQueryClosure: { + case ServeProto::Command::QueryClosure: { bool includeOutputs = readInt(in); StorePathSet closure; store->computeFSClosure(WorkerProto::Serialise::read(*store, in), @@ -960,7 +960,7 @@ static void opServe(Strings opFlags, Strings opArgs) break; } - case cmdAddToStoreNar: { + case ServeProto::Command::AddToStoreNar: { if (!writeAllowed) throw Error("importing paths is not allowed"); auto path = readString(in); From 9f69b7dee9fc6035b8aa0cc718f5e74af460d9aa Mon Sep 17 00:00:00 2001 From: John Ericson Date: Mon, 17 Apr 2023 13:40:46 -0400 Subject: [PATCH 4/5] Create `worker_proto::{Read,Write}Conn` Pass this around instead of `Source &` and `Sink &` directly. This will give us something to put the protocol version on once the time comes. To do this ergonomically, we need to expose `RemoteStore::Connection`, so do that too. Give it some more API docs while we are at it. --- src/libstore/build/derivation-goal.cc | 6 +- src/libstore/daemon.cc | 64 ++++++++------- src/libstore/derivations.cc | 7 +- src/libstore/export-import.cc | 7 +- src/libstore/legacy-ssh-store.cc | 52 ++++++++++-- src/libstore/path-info.cc | 7 +- src/libstore/remote-store-connection.hh | 97 ++++++++++++++++++++++ src/libstore/remote-store.cc | 76 +++++++++--------- src/libstore/remote-store.hh | 16 +--- src/libstore/ssh-store.cc | 1 + src/libstore/uds-remote-store.hh | 1 + src/libstore/worker-protocol-impl.hh | 40 +++++----- src/libstore/worker-protocol.cc | 102 ++++++++++++------------ src/libstore/worker-protocol.hh | 34 ++++++-- src/nix-store/nix-store.cc | 21 ++--- src/nix/daemon.cc | 1 + 16 files changed, 348 insertions(+), 184 deletions(-) create mode 100644 src/libstore/remote-store-connection.hh diff --git a/src/libstore/build/derivation-goal.cc b/src/libstore/build/derivation-goal.cc index 1a946e3d2..5e37f7ecb 100644 --- a/src/libstore/build/derivation-goal.cc +++ b/src/libstore/build/derivation-goal.cc @@ -1151,9 +1151,11 @@ HookReply DerivationGoal::tryBuildHook() throw; } + WorkerProto::WriteConn conn { hook->sink }; + /* Tell the hook all the inputs that have to be copied to the remote system. */ - WorkerProto::write(worker.store, hook->sink, inputPaths); + WorkerProto::write(worker.store, conn, inputPaths); /* Tell the hooks the missing outputs that have to be copied back from the remote system. */ @@ -1164,7 +1166,7 @@ HookReply DerivationGoal::tryBuildHook() if (buildMode != bmCheck && status.known && status.known->isValid()) continue; missingOutputs.insert(outputName); } - WorkerProto::write(worker.store, hook->sink, missingOutputs); + WorkerProto::write(worker.store, conn, missingOutputs); } hook->sink = FdSink(); diff --git a/src/libstore/daemon.cc b/src/libstore/daemon.cc index 7eba1a79d..75c3d2aca 100644 --- a/src/libstore/daemon.cc +++ b/src/libstore/daemon.cc @@ -260,13 +260,13 @@ struct ClientSettings } }; -static std::vector readDerivedPaths(Store & store, unsigned int clientVersion, Source & from) +static std::vector readDerivedPaths(Store & store, unsigned int clientVersion, WorkerProto::ReadConn conn) { std::vector reqs; if (GET_PROTOCOL_MINOR(clientVersion) >= 30) { - reqs = WorkerProto::Serialise>::read(store, from); + reqs = WorkerProto::Serialise>::read(store, conn); } else { - for (auto & s : readStrings(from)) + for (auto & s : readStrings(conn.from)) reqs.push_back(parsePathWithOutputs(store, s).toDerivedPath()); } return reqs; @@ -276,6 +276,9 @@ static void performOp(TunnelLogger * logger, ref store, TrustedFlag trusted, RecursiveFlag recursive, unsigned int clientVersion, Source & from, BufferedSink & to, WorkerProto::Op op) { + WorkerProto::ReadConn rconn { .from = from }; + WorkerProto::WriteConn wconn { .to = to }; + switch (op) { case WorkerProto::Op::IsValidPath: { @@ -288,7 +291,7 @@ static void performOp(TunnelLogger * logger, ref store, } case WorkerProto::Op::QueryValidPaths: { - auto paths = WorkerProto::Serialise::read(*store, from); + auto paths = WorkerProto::Serialise::read(*store, rconn); SubstituteFlag substitute = NoSubstitute; if (GET_PROTOCOL_MINOR(clientVersion) >= 27) { @@ -301,7 +304,7 @@ static void performOp(TunnelLogger * logger, ref store, } auto res = store->queryValidPaths(paths, substitute); logger->stopWork(); - WorkerProto::write(*store, to, res); + WorkerProto::write(*store, wconn, res); break; } @@ -317,11 +320,11 @@ static void performOp(TunnelLogger * logger, ref store, } case WorkerProto::Op::QuerySubstitutablePaths: { - auto paths = WorkerProto::Serialise::read(*store, from); + auto paths = WorkerProto::Serialise::read(*store, rconn); logger->startWork(); auto res = store->querySubstitutablePaths(paths); logger->stopWork(); - WorkerProto::write(*store, to, res); + WorkerProto::write(*store, wconn, res); break; } @@ -350,7 +353,7 @@ static void performOp(TunnelLogger * logger, ref store, paths = store->queryValidDerivers(path); else paths = store->queryDerivationOutputs(path); logger->stopWork(); - WorkerProto::write(*store, to, paths); + WorkerProto::write(*store, wconn, paths); break; } @@ -368,7 +371,7 @@ static void performOp(TunnelLogger * logger, ref store, logger->startWork(); auto outputs = store->queryPartialDerivationOutputMap(path); logger->stopWork(); - WorkerProto::write(*store, to, outputs); + WorkerProto::write(*store, wconn, outputs); break; } @@ -394,7 +397,7 @@ static void performOp(TunnelLogger * logger, ref store, if (GET_PROTOCOL_MINOR(clientVersion) >= 25) { auto name = readString(from); auto camStr = readString(from); - auto refs = WorkerProto::Serialise::read(*store, from); + auto refs = WorkerProto::Serialise::read(*store, rconn); bool repairBool; from >> repairBool; auto repair = RepairFlag{repairBool}; @@ -496,7 +499,7 @@ static void performOp(TunnelLogger * logger, ref store, case WorkerProto::Op::AddTextToStore: { std::string suffix = readString(from); std::string s = readString(from); - auto refs = WorkerProto::Serialise::read(*store, from); + auto refs = WorkerProto::Serialise::read(*store, rconn); logger->startWork(); auto path = store->addTextToStore(suffix, s, refs, NoRepair); logger->stopWork(); @@ -528,7 +531,7 @@ static void performOp(TunnelLogger * logger, ref store, } case WorkerProto::Op::BuildPaths: { - auto drvs = readDerivedPaths(*store, clientVersion, from); + auto drvs = readDerivedPaths(*store, clientVersion, rconn); BuildMode mode = bmNormal; if (GET_PROTOCOL_MINOR(clientVersion) >= 15) { mode = (BuildMode) readInt(from); @@ -553,7 +556,7 @@ static void performOp(TunnelLogger * logger, ref store, } case WorkerProto::Op::BuildPathsWithResults: { - auto drvs = readDerivedPaths(*store, clientVersion, from); + auto drvs = readDerivedPaths(*store, clientVersion, rconn); BuildMode mode = bmNormal; mode = (BuildMode) readInt(from); @@ -568,7 +571,7 @@ static void performOp(TunnelLogger * logger, ref store, auto results = store->buildPathsWithResults(drvs, mode); logger->stopWork(); - WorkerProto::write(*store, to, results); + WorkerProto::write(*store, wconn, results); break; } @@ -645,7 +648,7 @@ static void performOp(TunnelLogger * logger, ref store, DrvOutputs builtOutputs; for (auto & [output, realisation] : res.builtOutputs) builtOutputs.insert_or_assign(realisation.id, realisation); - WorkerProto::write(*store, to, builtOutputs); + WorkerProto::write(*store, wconn, builtOutputs); } break; } @@ -710,7 +713,7 @@ static void performOp(TunnelLogger * logger, ref store, case WorkerProto::Op::CollectGarbage: { GCOptions options; options.action = (GCOptions::GCAction) readInt(from); - options.pathsToDelete = WorkerProto::Serialise::read(*store, from); + options.pathsToDelete = WorkerProto::Serialise::read(*store, rconn); from >> options.ignoreLiveness >> options.maxFreed; // obsolete fields readInt(from); @@ -780,7 +783,7 @@ static void performOp(TunnelLogger * logger, ref store, else { to << 1 << (i->second.deriver ? store->printStorePath(*i->second.deriver) : ""); - WorkerProto::write(*store, to, i->second.references); + WorkerProto::write(*store, wconn, i->second.references); to << i->second.downloadSize << i->second.narSize; } @@ -791,11 +794,11 @@ static void performOp(TunnelLogger * logger, ref store, SubstitutablePathInfos infos; StorePathCAMap pathsMap = {}; if (GET_PROTOCOL_MINOR(clientVersion) < 22) { - auto paths = WorkerProto::Serialise::read(*store, from); + auto paths = WorkerProto::Serialise::read(*store, rconn); for (auto & path : paths) pathsMap.emplace(path, std::nullopt); } else - pathsMap = WorkerProto::Serialise::read(*store, from); + pathsMap = WorkerProto::Serialise::read(*store, rconn); logger->startWork(); store->querySubstitutablePathInfos(pathsMap, infos); logger->stopWork(); @@ -803,7 +806,7 @@ static void performOp(TunnelLogger * logger, ref store, for (auto & i : infos) { to << store->printStorePath(i.first) << (i.second.deriver ? store->printStorePath(*i.second.deriver) : ""); - WorkerProto::write(*store, to, i.second.references); + WorkerProto::write(*store, wconn, i.second.references); to << i.second.downloadSize << i.second.narSize; } break; @@ -813,7 +816,7 @@ static void performOp(TunnelLogger * logger, ref store, logger->startWork(); auto paths = store->queryAllValidPaths(); logger->stopWork(); - WorkerProto::write(*store, to, paths); + WorkerProto::write(*store, wconn, paths); break; } @@ -885,7 +888,7 @@ static void performOp(TunnelLogger * logger, ref store, ValidPathInfo info { path, narHash }; if (deriver != "") info.deriver = store->parseStorePath(deriver); - info.references = WorkerProto::Serialise::read(*store, from); + info.references = WorkerProto::Serialise::read(*store, rconn); from >> info.registrationTime >> info.narSize >> info.ultimate; info.sigs = readStrings(from); info.ca = ContentAddress::parseOpt(readString(from)); @@ -930,15 +933,15 @@ static void performOp(TunnelLogger * logger, ref store, } case WorkerProto::Op::QueryMissing: { - auto targets = readDerivedPaths(*store, clientVersion, from); + auto targets = readDerivedPaths(*store, clientVersion, rconn); logger->startWork(); StorePathSet willBuild, willSubstitute, unknown; uint64_t downloadSize, narSize; store->queryMissing(targets, willBuild, willSubstitute, unknown, downloadSize, narSize); logger->stopWork(); - WorkerProto::write(*store, to, willBuild); - WorkerProto::write(*store, to, willSubstitute); - WorkerProto::write(*store, to, unknown); + WorkerProto::write(*store, wconn, willBuild); + WorkerProto::write(*store, wconn, willSubstitute); + WorkerProto::write(*store, wconn, unknown); to << downloadSize << narSize; break; } @@ -951,7 +954,7 @@ static void performOp(TunnelLogger * logger, ref store, store->registerDrvOutput(Realisation{ .id = outputId, .outPath = outputPath}); } else { - auto realisation = WorkerProto::Serialise::read(*store, from); + auto realisation = WorkerProto::Serialise::read(*store, rconn); store->registerDrvOutput(realisation); } logger->stopWork(); @@ -966,11 +969,11 @@ static void performOp(TunnelLogger * logger, ref store, if (GET_PROTOCOL_MINOR(clientVersion) < 31) { std::set outPaths; if (info) outPaths.insert(info->outPath); - WorkerProto::write(*store, to, outPaths); + WorkerProto::write(*store, wconn, outPaths); } else { std::set realisations; if (info) realisations.insert(*info); - WorkerProto::write(*store, to, realisations); + WorkerProto::write(*store, wconn, realisations); } break; } @@ -1050,7 +1053,8 @@ void processConnection( auto temp = trusted ? store->isTrustedClient() : std::optional { NotTrusted }; - WorkerProto::write(*store, to, temp); + WorkerProto::WriteConn wconn { .to = to }; + WorkerProto::write(*store, wconn, temp); } /* Send startup error messages to the client. */ diff --git a/src/libstore/derivations.cc b/src/libstore/derivations.cc index 2295ff9e8..6f63685d4 100644 --- a/src/libstore/derivations.cc +++ b/src/libstore/derivations.cc @@ -750,7 +750,8 @@ Source & readDerivation(Source & in, const Store & store, BasicDerivation & drv, drv.outputs.emplace(std::move(name), std::move(output)); } - drv.inputSrcs = WorkerProto::Serialise::read(store, in); + drv.inputSrcs = WorkerProto::Serialise::read(store, + WorkerProto::ReadConn { .from = in }); in >> drv.platform >> drv.builder; drv.args = readStrings(in); @@ -798,7 +799,9 @@ void writeDerivation(Sink & out, const Store & store, const BasicDerivation & dr }, }, i.second.raw()); } - WorkerProto::write(store, out, drv.inputSrcs); + WorkerProto::write(store, + WorkerProto::WriteConn { .to = out }, + drv.inputSrcs); out << drv.platform << drv.builder << drv.args; out << drv.env.size(); for (auto & i : drv.env) diff --git a/src/libstore/export-import.cc b/src/libstore/export-import.cc index b4100bace..e866aeb42 100644 --- a/src/libstore/export-import.cc +++ b/src/libstore/export-import.cc @@ -46,7 +46,9 @@ void Store::exportPath(const StorePath & path, Sink & sink) teeSink << exportMagic << printStorePath(path); - WorkerProto::write(*this, teeSink, info->references); + WorkerProto::write(*this, + WorkerProto::WriteConn { .to = teeSink }, + info->references); teeSink << (info->deriver ? printStorePath(*info->deriver) : "") << 0; @@ -74,7 +76,8 @@ StorePaths Store::importPaths(Source & source, CheckSigsFlag checkSigs) //Activity act(*logger, lvlInfo, "importing path '%s'", info.path); - auto references = WorkerProto::Serialise::read(*this, source); + auto references = WorkerProto::Serialise::read(*this, + WorkerProto::ReadConn { .from = source }); auto deriver = readString(source); auto narHash = hashString(htSHA256, saved.s); diff --git a/src/libstore/legacy-ssh-store.cc b/src/libstore/legacy-ssh-store.cc index 55ecab0ff..fa17d606d 100644 --- a/src/libstore/legacy-ssh-store.cc +++ b/src/libstore/legacy-ssh-store.cc @@ -48,6 +48,42 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor FdSource from; int remoteVersion; bool good = true; + + /** + * Coercion to `WorkerProto::ReadConn`. This makes it easy to use the + * factored out worker protocol searlizers with a + * `LegacySSHStore::Connection`. + * + * The worker protocol connection types are unidirectional, unlike + * this type. + * + * @todo Use server protocol serializers, not worker protocol + * serializers, once we have made that distiction. + */ + operator WorkerProto::ReadConn () + { + return WorkerProto::ReadConn { + .from = from, + }; + } + + /* + * Coercion to `WorkerProto::WriteConn`. This makes it easy to use the + * factored out worker protocol searlizers with a + * `LegacySSHStore::Connection`. + * + * The worker protocol connection types are unidirectional, unlike + * this type. + * + * @todo Use server protocol serializers, not worker protocol + * serializers, once we have made that distiction. + */ + operator WorkerProto::WriteConn () + { + return WorkerProto::WriteConn { + .to = to, + }; + } }; std::string host; @@ -147,7 +183,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor auto deriver = readString(conn->from); if (deriver != "") info->deriver = parseStorePath(deriver); - info->references = WorkerProto::Serialise::read(*this, conn->from); + info->references = WorkerProto::Serialise::read(*this, *conn); readLongLong(conn->from); // download size info->narSize = readLongLong(conn->from); @@ -181,7 +217,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor << printStorePath(info.path) << (info.deriver ? printStorePath(*info.deriver) : "") << info.narHash.to_string(Base16, false); - WorkerProto::write(*this, conn->to, info.references); + WorkerProto::write(*this, *conn, info.references); conn->to << info.registrationTime << info.narSize @@ -210,7 +246,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor conn->to << exportMagic << printStorePath(info.path); - WorkerProto::write(*this, conn->to, info.references); + WorkerProto::write(*this, *conn, info.references); conn->to << (info.deriver ? printStorePath(*info.deriver) : "") << 0 @@ -295,7 +331,7 @@ public: if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 3) conn->from >> status.timesBuilt >> status.isNonDeterministic >> status.startTime >> status.stopTime; if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 6) { - auto builtOutputs = WorkerProto::Serialise::read(*this, conn->from); + auto builtOutputs = WorkerProto::Serialise::read(*this, *conn); for (auto && [output, realisation] : builtOutputs) status.builtOutputs.insert_or_assign( std::move(output.outputName), @@ -370,10 +406,10 @@ public: conn->to << ServeProto::Command::QueryClosure << includeOutputs; - WorkerProto::write(*this, conn->to, paths); + WorkerProto::write(*this, *conn, paths); conn->to.flush(); - for (auto & i : WorkerProto::Serialise::read(*this, conn->from)) + for (auto & i : WorkerProto::Serialise::read(*this, *conn)) out.insert(i); } @@ -386,10 +422,10 @@ public: << ServeProto::Command::QueryValidPaths << false // lock << maybeSubstitute; - WorkerProto::write(*this, conn->to, paths); + WorkerProto::write(*this, *conn, paths); conn->to.flush(); - return WorkerProto::Serialise::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, *conn); } void connect() override diff --git a/src/libstore/path-info.cc b/src/libstore/path-info.cc index 9971de1dd..981bbfb14 100644 --- a/src/libstore/path-info.cc +++ b/src/libstore/path-info.cc @@ -133,7 +133,8 @@ ValidPathInfo ValidPathInfo::read(Source & source, const Store & store, unsigned auto narHash = Hash::parseAny(readString(source), htSHA256); ValidPathInfo info(path, narHash); if (deriver != "") info.deriver = store.parseStorePath(deriver); - info.references = WorkerProto::Serialise::read(store, source); + info.references = WorkerProto::Serialise::read(store, + WorkerProto::ReadConn { .from = source }); source >> info.registrationTime >> info.narSize; if (format >= 16) { source >> info.ultimate; @@ -154,7 +155,9 @@ void ValidPathInfo::write( sink << store.printStorePath(path); sink << (deriver ? store.printStorePath(*deriver) : "") << narHash.to_string(Base16, false); - WorkerProto::write(store, sink, references); + WorkerProto::write(store, + WorkerProto::WriteConn { .to = sink }, + references); sink << registrationTime << narSize; if (format >= 16) { sink << ultimate diff --git a/src/libstore/remote-store-connection.hh b/src/libstore/remote-store-connection.hh new file mode 100644 index 000000000..d32d91a60 --- /dev/null +++ b/src/libstore/remote-store-connection.hh @@ -0,0 +1,97 @@ +#include "remote-store.hh" +#include "worker-protocol.hh" + +namespace nix { + +/** + * Bidirectional connection (send and receive) used by the Remote Store + * implementation. + * + * Contains `Source` and `Sink` for actual communication, along with + * other information learned when negotiating the connection. + */ +struct RemoteStore::Connection +{ + /** + * Send with this. + */ + FdSink to; + + /** + * Receive with this. + */ + FdSource from; + + /** + * Worker protocol version used for the connection. + * + * Despite its name, I think it is actually the maximum version both + * sides support. (If the maximum doesn't exist, we would fail to + * establish a connection and produce a value of this type.) + */ + unsigned int daemonVersion; + + /** + * Whether the remote side trusts us or not. + * + * 3 values: "yes", "no", or `std::nullopt` for "unknown". + * + * Note that the "remote side" might not be just the end daemon, but + * also an intermediary forwarder that can make its own trusting + * decisions. This would be the intersection of all their trust + * decisions, since it takes only one link in the chain to start + * denying operations. + */ + std::optional remoteTrustsUs; + + /** + * The version of the Nix daemon that is processing our requests. + * + * Do note, it may or may not communicating with another daemon, + * rather than being an "end" `LocalStore` or similar. + */ + std::optional daemonNixVersion; + + /** + * Time this connection was established. + */ + std::chrono::time_point startTime; + + /** + * Coercion to `WorkerProto::ReadConn`. This makes it easy to use the + * factored out worker protocol searlizers with a + * `RemoteStore::Connection`. + * + * The worker protocol connection types are unidirectional, unlike + * this type. + */ + operator WorkerProto::ReadConn () + { + return WorkerProto::ReadConn { + .from = from, + }; + } + + /** + * Coercion to `WorkerProto::WriteConn`. This makes it easy to use the + * factored out worker protocol searlizers with a + * `RemoteStore::Connection`. + * + * The worker protocol connection types are unidirectional, unlike + * this type. + */ + operator WorkerProto::WriteConn () + { + return WorkerProto::WriteConn { + .to = to, + }; + } + + virtual ~Connection(); + + virtual void closeWrite() = 0; + + std::exception_ptr processStderr(Sink * sink = 0, Source * source = 0, bool flush = true); +}; + +} diff --git a/src/libstore/remote-store.cc b/src/libstore/remote-store.cc index 9253ce09e..1e2104e1f 100644 --- a/src/libstore/remote-store.cc +++ b/src/libstore/remote-store.cc @@ -5,6 +5,7 @@ #include "remote-fs-accessor.hh" #include "build-result.hh" #include "remote-store.hh" +#include "remote-store-connection.hh" #include "worker-protocol.hh" #include "worker-protocol-impl.hh" #include "archive.hh" @@ -101,7 +102,7 @@ void RemoteStore::initConnection(Connection & conn) } if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 35) { - conn.remoteTrustsUs = WorkerProto::Serialise>::read(*this, conn.from); + conn.remoteTrustsUs = WorkerProto::Serialise>::read(*this, conn); } else { // We don't know the answer; protocol to old. conn.remoteTrustsUs = std::nullopt; @@ -185,6 +186,7 @@ struct ConnectionHandle } RemoteStore::Connection * operator -> () { return &*handle; } + RemoteStore::Connection & operator * () { return *handle; } void processStderr(Sink * sink = 0, Source * source = 0, bool flush = true) { @@ -228,12 +230,12 @@ StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, Substitute return res; } else { conn->to << WorkerProto::Op::QueryValidPaths; - WorkerProto::write(*this, conn->to, paths); + WorkerProto::write(*this, *conn, paths); if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 27) { conn->to << (settings.buildersUseSubstitutes ? 1 : 0); } conn.processStderr(); - return WorkerProto::Serialise::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, *conn); } } @@ -243,7 +245,7 @@ StorePathSet RemoteStore::queryAllValidPaths() auto conn(getConnection()); conn->to << WorkerProto::Op::QueryAllValidPaths; conn.processStderr(); - return WorkerProto::Serialise::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, *conn); } @@ -260,9 +262,9 @@ StorePathSet RemoteStore::querySubstitutablePaths(const StorePathSet & paths) return res; } else { conn->to << WorkerProto::Op::QuerySubstitutablePaths; - WorkerProto::write(*this, conn->to, paths); + WorkerProto::write(*this, *conn, paths); conn.processStderr(); - return WorkerProto::Serialise::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, *conn); } } @@ -284,7 +286,7 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S auto deriver = readString(conn->from); if (deriver != "") info.deriver = parseStorePath(deriver); - info.references = WorkerProto::Serialise::read(*this, conn->from); + info.references = WorkerProto::Serialise::read(*this, *conn); info.downloadSize = readLongLong(conn->from); info.narSize = readLongLong(conn->from); infos.insert_or_assign(i.first, std::move(info)); @@ -297,9 +299,9 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S StorePathSet paths; for (auto & path : pathsMap) paths.insert(path.first); - WorkerProto::write(*this, conn->to, paths); + WorkerProto::write(*this, *conn, paths); } else - WorkerProto::write(*this, conn->to, pathsMap); + WorkerProto::write(*this, *conn, pathsMap); conn.processStderr(); size_t count = readNum(conn->from); for (size_t n = 0; n < count; n++) { @@ -307,7 +309,7 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S auto deriver = readString(conn->from); if (deriver != "") info.deriver = parseStorePath(deriver); - info.references = WorkerProto::Serialise::read(*this, conn->from); + info.references = WorkerProto::Serialise::read(*this, *conn); info.downloadSize = readLongLong(conn->from); info.narSize = readLongLong(conn->from); } @@ -350,7 +352,7 @@ void RemoteStore::queryReferrers(const StorePath & path, auto conn(getConnection()); conn->to << WorkerProto::Op::QueryReferrers << printStorePath(path); conn.processStderr(); - for (auto & i : WorkerProto::Serialise::read(*this, conn->from)) + for (auto & i : WorkerProto::Serialise::read(*this, *conn)) referrers.insert(i); } @@ -360,7 +362,7 @@ StorePathSet RemoteStore::queryValidDerivers(const StorePath & path) auto conn(getConnection()); conn->to << WorkerProto::Op::QueryValidDerivers << printStorePath(path); conn.processStderr(); - return WorkerProto::Serialise::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, *conn); } @@ -372,7 +374,7 @@ StorePathSet RemoteStore::queryDerivationOutputs(const StorePath & path) auto conn(getConnection()); conn->to << WorkerProto::Op::QueryDerivationOutputs << printStorePath(path); conn.processStderr(); - return WorkerProto::Serialise::read(*this, conn->from); + return WorkerProto::Serialise::read(*this, *conn); } @@ -382,7 +384,7 @@ std::map> RemoteStore::queryPartialDerivat auto conn(getConnection()); conn->to << WorkerProto::Op::QueryDerivationOutputMap << printStorePath(path); conn.processStderr(); - return WorkerProto::Serialise>>::read(*this, conn->from); + return WorkerProto::Serialise>>::read(*this, *conn); } else { // Fallback for old daemon versions. // For floating-CA derivations (and their co-dependencies) this is an @@ -428,7 +430,7 @@ ref RemoteStore::addCAToStore( << WorkerProto::Op::AddToStore << name << caMethod.render(hashType); - WorkerProto::write(*this, conn->to, references); + WorkerProto::write(*this, *conn, references); conn->to << repair; // The dump source may invoke the store, so we need to make some room. @@ -453,7 +455,7 @@ ref RemoteStore::addCAToStore( name, printHashType(hashType)); std::string s = dump.drain(); conn->to << WorkerProto::Op::AddTextToStore << name << s; - WorkerProto::write(*this, conn->to, references); + WorkerProto::write(*this, *conn, references); conn.processStderr(); }, [&](const FileIngestionMethod & fim) -> void { @@ -519,7 +521,7 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source, sink << exportMagic << printStorePath(info.path); - WorkerProto::write(*this, sink, info.references); + WorkerProto::write(*this, *conn, info.references); sink << (info.deriver ? printStorePath(*info.deriver) : "") << 0 // == no legacy signature @@ -529,7 +531,7 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source, conn.processStderr(0, source2.get()); - auto importedPaths = WorkerProto::Serialise::read(*this, conn->from); + auto importedPaths = WorkerProto::Serialise::read(*this, *conn); assert(importedPaths.size() <= 1); } @@ -538,7 +540,7 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source, << printStorePath(info.path) << (info.deriver ? printStorePath(*info.deriver) : "") << info.narHash.to_string(Base16, false); - WorkerProto::write(*this, conn->to, info.references); + WorkerProto::write(*this, *conn, info.references); conn->to << info.registrationTime << info.narSize << info.ultimate << info.sigs << renderContentAddress(info.ca) << repair << !checkSigs; @@ -611,7 +613,7 @@ void RemoteStore::registerDrvOutput(const Realisation & info) conn->to << info.id.to_string(); conn->to << std::string(info.outPath.to_string()); } else { - WorkerProto::write(*this, conn->to, info); + WorkerProto::write(*this, *conn, info); } conn.processStderr(); } @@ -634,13 +636,13 @@ void RemoteStore::queryRealisationUncached(const DrvOutput & id, auto real = [&]() -> std::shared_ptr { if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) { auto outPaths = WorkerProto::Serialise>::read( - *this, conn->from); + *this, *conn); if (outPaths.empty()) return nullptr; return std::make_shared(Realisation { .id = id, .outPath = *outPaths.begin() }); } else { auto realisations = WorkerProto::Serialise>::read( - *this, conn->from); + *this, *conn); if (realisations.empty()) return nullptr; return std::make_shared(*realisations.begin()); @@ -651,10 +653,10 @@ void RemoteStore::queryRealisationUncached(const DrvOutput & id, } catch (...) { return callback.rethrow(); } } -static void writeDerivedPaths(RemoteStore & store, ConnectionHandle & conn, const std::vector & reqs) +static void writeDerivedPaths(RemoteStore & store, RemoteStore::Connection & conn, const std::vector & reqs) { - if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 30) { - WorkerProto::write(store, conn->to, reqs); + if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 30) { + WorkerProto::write(store, conn, reqs); } else { Strings ss; for (auto & p : reqs) { @@ -666,12 +668,12 @@ static void writeDerivedPaths(RemoteStore & store, ConnectionHandle & conn, cons [&](const StorePath & drvPath) { throw Error("trying to request '%s', but daemon protocol %d.%d is too old (< 1.29) to request a derivation file", store.printStorePath(drvPath), - GET_PROTOCOL_MAJOR(conn->daemonVersion), - GET_PROTOCOL_MINOR(conn->daemonVersion)); + GET_PROTOCOL_MAJOR(conn.daemonVersion), + GET_PROTOCOL_MINOR(conn.daemonVersion)); }, }, sOrDrvPath); } - conn->to << ss; + conn.to << ss; } } @@ -697,7 +699,7 @@ void RemoteStore::buildPaths(const std::vector & drvPaths, BuildMod auto conn(getConnection()); conn->to << WorkerProto::Op::BuildPaths; assert(GET_PROTOCOL_MINOR(conn->daemonVersion) >= 13); - writeDerivedPaths(*this, conn, drvPaths); + writeDerivedPaths(*this, *conn, drvPaths); if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 15) conn->to << buildMode; else @@ -721,10 +723,10 @@ std::vector RemoteStore::buildPathsWithResults( if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) { conn->to << WorkerProto::Op::BuildPathsWithResults; - writeDerivedPaths(*this, conn, paths); + writeDerivedPaths(*this, *conn, paths); conn->to << buildMode; conn.processStderr(); - return WorkerProto::Serialise>::read(*this, conn->from); + return WorkerProto::Serialise>::read(*this, *conn); } else { // Avoid deadlock. conn_.reset(); @@ -807,7 +809,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::Serialise::read(*this, conn->from); + auto builtOutputs = WorkerProto::Serialise::read(*this, *conn); for (auto && [output, realisation] : builtOutputs) res.builtOutputs.insert_or_assign( std::move(output.outputName), @@ -866,7 +868,7 @@ void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results) conn->to << WorkerProto::Op::CollectGarbage << options.action; - WorkerProto::write(*this, conn->to, options.pathsToDelete); + WorkerProto::write(*this, *conn, options.pathsToDelete); conn->to << options.ignoreLiveness << options.maxFreed /* removed options */ @@ -923,11 +925,11 @@ void RemoteStore::queryMissing(const std::vector & targets, // to prevent a deadlock. goto fallback; conn->to << WorkerProto::Op::QueryMissing; - writeDerivedPaths(*this, conn, targets); + writeDerivedPaths(*this, *conn, targets); conn.processStderr(); - willBuild = WorkerProto::Serialise::read(*this, conn->from); - willSubstitute = WorkerProto::Serialise::read(*this, conn->from); - unknown = WorkerProto::Serialise::read(*this, conn->from); + willBuild = WorkerProto::Serialise::read(*this, *conn); + willSubstitute = WorkerProto::Serialise::read(*this, *conn); + unknown = WorkerProto::Serialise::read(*this, *conn); conn->from >> downloadSize >> narSize; return; } diff --git a/src/libstore/remote-store.hh b/src/libstore/remote-store.hh index 4f3971bfd..cb7a71acf 100644 --- a/src/libstore/remote-store.hh +++ b/src/libstore/remote-store.hh @@ -166,21 +166,7 @@ public: void flushBadConnections(); - struct Connection - { - FdSink to; - FdSource from; - unsigned int daemonVersion; - std::optional remoteTrustsUs; - std::optional daemonNixVersion; - std::chrono::time_point startTime; - - virtual ~Connection(); - - virtual void closeWrite() = 0; - - std::exception_ptr processStderr(Sink * sink = 0, Source * source = 0, bool flush = true); - }; + struct Connection; ref openConnectionWrapper(); diff --git a/src/libstore/ssh-store.cc b/src/libstore/ssh-store.cc index 962221ad2..0200076c0 100644 --- a/src/libstore/ssh-store.cc +++ b/src/libstore/ssh-store.cc @@ -1,6 +1,7 @@ #include "ssh-store-config.hh" #include "store-api.hh" #include "remote-store.hh" +#include "remote-store-connection.hh" #include "remote-fs-accessor.hh" #include "archive.hh" #include "worker-protocol.hh" diff --git a/src/libstore/uds-remote-store.hh b/src/libstore/uds-remote-store.hh index 3bbab371c..2bd6517fa 100644 --- a/src/libstore/uds-remote-store.hh +++ b/src/libstore/uds-remote-store.hh @@ -2,6 +2,7 @@ ///@file #include "remote-store.hh" +#include "remote-store-connection.hh" #include "local-fs-store.hh" namespace nix { diff --git a/src/libstore/worker-protocol-impl.hh b/src/libstore/worker-protocol-impl.hh index f34e55c54..d3d2792ff 100644 --- a/src/libstore/worker-protocol-impl.hh +++ b/src/libstore/worker-protocol-impl.hh @@ -13,65 +13,65 @@ namespace nix { template -std::vector WorkerProto::Serialise>::read(const Store & store, Source & from) +std::vector WorkerProto::Serialise>::read(const Store & store, WorkerProto::ReadConn conn) { std::vector resSet; - auto size = readNum(from); + auto size = readNum(conn.from); while (size--) { - resSet.push_back(WorkerProto::Serialise::read(store, from)); + resSet.push_back(WorkerProto::Serialise::read(store, conn)); } return resSet; } template -void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::vector & resSet) +void WorkerProto::Serialise>::write(const Store & store, WorkerProto::WriteConn conn, const std::vector & resSet) { - out << resSet.size(); + conn.to << resSet.size(); for (auto & key : resSet) { - WorkerProto::Serialise::write(store, out, key); + WorkerProto::Serialise::write(store, conn, key); } } template -std::set WorkerProto::Serialise>::read(const Store & store, Source & from) +std::set WorkerProto::Serialise>::read(const Store & store, WorkerProto::ReadConn conn) { std::set resSet; - auto size = readNum(from); + auto size = readNum(conn.from); while (size--) { - resSet.insert(WorkerProto::Serialise::read(store, from)); + resSet.insert(WorkerProto::Serialise::read(store, conn)); } return resSet; } template -void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::set & resSet) +void WorkerProto::Serialise>::write(const Store & store, WorkerProto::WriteConn conn, const std::set & resSet) { - out << resSet.size(); + conn.to << resSet.size(); for (auto & key : resSet) { - WorkerProto::Serialise::write(store, out, key); + WorkerProto::Serialise::write(store, conn, key); } } template -std::map WorkerProto::Serialise>::read(const Store & store, Source & from) +std::map WorkerProto::Serialise>::read(const Store & store, WorkerProto::ReadConn conn) { std::map resMap; - auto size = readNum(from); + auto size = readNum(conn.from); while (size--) { - auto k = WorkerProto::Serialise::read(store, from); - auto v = WorkerProto::Serialise::read(store, from); + auto k = WorkerProto::Serialise::read(store, conn); + auto v = WorkerProto::Serialise::read(store, conn); resMap.insert_or_assign(std::move(k), std::move(v)); } return resMap; } template -void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::map & resMap) +void WorkerProto::Serialise>::write(const Store & store, WorkerProto::WriteConn conn, const std::map & resMap) { - out << resMap.size(); + conn.to << resMap.size(); for (auto & i : resMap) { - WorkerProto::Serialise::write(store, out, i.first); - WorkerProto::Serialise::write(store, out, i.second); + WorkerProto::Serialise::write(store, conn, i.first); + WorkerProto::Serialise::write(store, conn, i.second); } } diff --git a/src/libstore/worker-protocol.cc b/src/libstore/worker-protocol.cc index 0bf9e4d68..a23130743 100644 --- a/src/libstore/worker-protocol.cc +++ b/src/libstore/worker-protocol.cc @@ -12,31 +12,31 @@ namespace nix { -std::string WorkerProto::Serialise::read(const Store & store, Source & from) +std::string WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { - return readString(from); + return readString(conn.from); } -void WorkerProto::Serialise::write(const Store & store, Sink & out, const std::string & str) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const std::string & str) { - out << str; + conn.to << str; } -StorePath WorkerProto::Serialise::read(const Store & store, Source & from) +StorePath WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { - return store.parseStorePath(readString(from)); + return store.parseStorePath(readString(conn.from)); } -void WorkerProto::Serialise::write(const Store & store, Sink & out, const StorePath & storePath) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const StorePath & storePath) { - out << store.printStorePath(storePath); + conn.to << store.printStorePath(storePath); } -std::optional WorkerProto::Serialise>::read(const Store & store, Source & from) +std::optional WorkerProto::Serialise>::read(const Store & store, WorkerProto::ReadConn conn) { - auto temp = readNum(from); + auto temp = readNum(conn.from); switch (temp) { case 0: return std::nullopt; @@ -49,17 +49,17 @@ std::optional WorkerProto::Serialise>::r } } -void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::optional & optTrusted) +void WorkerProto::Serialise>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional & optTrusted) { if (!optTrusted) - out << (uint8_t)0; + conn.to << (uint8_t)0; else { switch (*optTrusted) { case Trusted: - out << (uint8_t)1; + conn.to << (uint8_t)1; break; case NotTrusted: - out << (uint8_t)2; + conn.to << (uint8_t)2; break; default: assert(false); @@ -68,83 +68,83 @@ void WorkerProto::Serialise>::write(const Store & sto } -ContentAddress WorkerProto::Serialise::read(const Store & store, Source & from) +ContentAddress WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { - return ContentAddress::parse(readString(from)); + return ContentAddress::parse(readString(conn.from)); } -void WorkerProto::Serialise::write(const Store & store, Sink & out, const ContentAddress & ca) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const ContentAddress & ca) { - out << renderContentAddress(ca); + conn.to << renderContentAddress(ca); } -DerivedPath WorkerProto::Serialise::read(const Store & store, Source & from) +DerivedPath WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { - auto s = readString(from); + auto s = readString(conn.from); return DerivedPath::parseLegacy(store, s); } -void WorkerProto::Serialise::write(const Store & store, Sink & out, const DerivedPath & req) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const DerivedPath & req) { - out << req.to_string_legacy(store); + conn.to << req.to_string_legacy(store); } -Realisation WorkerProto::Serialise::read(const Store & store, Source & from) +Realisation WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { - std::string rawInput = readString(from); + std::string rawInput = readString(conn.from); return Realisation::fromJSON( nlohmann::json::parse(rawInput), "remote-protocol" ); } -void WorkerProto::Serialise::write(const Store & store, Sink & out, const Realisation & realisation) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const Realisation & realisation) { - out << realisation.toJSON().dump(); + conn.to << realisation.toJSON().dump(); } -DrvOutput WorkerProto::Serialise::read(const Store & store, Source & from) +DrvOutput WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { - return DrvOutput::parse(readString(from)); + return DrvOutput::parse(readString(conn.from)); } -void WorkerProto::Serialise::write(const Store & store, Sink & out, const DrvOutput & drvOutput) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const DrvOutput & drvOutput) { - out << drvOutput.to_string(); + conn.to << drvOutput.to_string(); } -KeyedBuildResult WorkerProto::Serialise::read(const Store & store, Source & from) +KeyedBuildResult WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { - auto path = WorkerProto::Serialise::read(store, from); - auto br = WorkerProto::Serialise::read(store, from); + auto path = WorkerProto::Serialise::read(store, conn); + auto br = WorkerProto::Serialise::read(store, conn); return KeyedBuildResult { std::move(br), /* .path = */ std::move(path), }; } -void WorkerProto::Serialise::write(const Store & store, Sink & to, const KeyedBuildResult & res) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const KeyedBuildResult & res) { - WorkerProto::write(store, to, res.path); - WorkerProto::write(store, to, static_cast(res)); + WorkerProto::write(store, conn, res.path); + WorkerProto::write(store, conn, static_cast(res)); } -BuildResult WorkerProto::Serialise::read(const Store & store, Source & from) +BuildResult WorkerProto::Serialise::read(const Store & store, WorkerProto::ReadConn conn) { BuildResult res; - res.status = (BuildResult::Status) readInt(from); - from + res.status = (BuildResult::Status) readInt(conn.from); + conn.from >> res.errorMsg >> res.timesBuilt >> res.isNonDeterministic >> res.startTime >> res.stopTime; - auto builtOutputs = WorkerProto::Serialise::read(store, from); + auto builtOutputs = WorkerProto::Serialise::read(store, conn); for (auto && [output, realisation] : builtOutputs) res.builtOutputs.insert_or_assign( std::move(output.outputName), @@ -152,9 +152,9 @@ BuildResult WorkerProto::Serialise::read(const Store & store, Sourc return res; } -void WorkerProto::Serialise::write(const Store & store, Sink & to, const BuildResult & res) +void WorkerProto::Serialise::write(const Store & store, WorkerProto::WriteConn conn, const BuildResult & res) { - to + conn.to << res.status << res.errorMsg << res.timesBuilt @@ -164,30 +164,30 @@ void WorkerProto::Serialise::write(const Store & store, Sink & to, DrvOutputs builtOutputs; for (auto & [output, realisation] : res.builtOutputs) builtOutputs.insert_or_assign(realisation.id, realisation); - WorkerProto::write(store, to, builtOutputs); + WorkerProto::write(store, conn, builtOutputs); } -std::optional WorkerProto::Serialise>::read(const Store & store, Source & from) +std::optional WorkerProto::Serialise>::read(const Store & store, WorkerProto::ReadConn conn) { - auto s = readString(from); + auto s = readString(conn.from); return s == "" ? std::optional {} : store.parseStorePath(s); } -void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::optional & storePathOpt) +void WorkerProto::Serialise>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional & storePathOpt) { - out << (storePathOpt ? store.printStorePath(*storePathOpt) : ""); + conn.to << (storePathOpt ? store.printStorePath(*storePathOpt) : ""); } -std::optional WorkerProto::Serialise>::read(const Store & store, Source & from) +std::optional WorkerProto::Serialise>::read(const Store & store, WorkerProto::ReadConn conn) { - return ContentAddress::parseOpt(readString(from)); + return ContentAddress::parseOpt(readString(conn.from)); } -void WorkerProto::Serialise>::write(const Store & store, Sink & out, const std::optional & caOpt) +void WorkerProto::Serialise>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional & caOpt) { - out << (caOpt ? renderContentAddress(*caOpt) : ""); + conn.to << (caOpt ? renderContentAddress(*caOpt) : ""); } } diff --git a/src/libstore/worker-protocol.hh b/src/libstore/worker-protocol.hh index cd6801290..ff762c924 100644 --- a/src/libstore/worker-protocol.hh +++ b/src/libstore/worker-protocol.hh @@ -49,6 +49,28 @@ struct WorkerProto */ enum struct Op : uint64_t; + /** + * A unidirectional read connection, to be used by the read half of the + * canonical serializers below. + * + * This currently is just a `Source &`, but more fields will be added + * later. + */ + struct ReadConn { + Source & from; + }; + + /** + * A unidirectional write connection, to be used by the write half of the + * canonical serializers below. + * + * This currently is just a `Sink &`, but more fields will be added + * later. + */ + struct WriteConn { + Sink & to; + }; + /** * Data type for canonical pairs of serialisers for the worker protocol. * @@ -75,8 +97,8 @@ struct WorkerProto // This makes for a quicker debug cycle, as desired. #if 0 { - static T read(const Store & store, Source & from); - static void write(const Store & store, Sink & out, const T & t); + static T read(const Store & store, ReadConn conn); + static void write(const Store & store, WriteConn conn, const T & t); }; #endif @@ -85,9 +107,9 @@ struct WorkerProto * infer the type instead of having to write it down explicitly. */ template - static void write(const Store & store, Sink & out, const T & t) + static void write(const Store & store, WriteConn conn, const T & t) { - WorkerProto::Serialise::write(store, out, t); + WorkerProto::Serialise::write(store, conn, t); } }; @@ -171,8 +193,8 @@ inline std::ostream & operator << (std::ostream & s, WorkerProto::Op op) */ #define MAKE_WORKER_PROTO(T) \ struct WorkerProto::Serialise< T > { \ - static T read(const Store & store, Source & from); \ - static void write(const Store & store, Sink & out, const T & t); \ + static T read(const Store & store, WorkerProto::ReadConn conn); \ + static void write(const Store & store, WorkerProto::WriteConn conn, const T & t); \ }; template<> diff --git a/src/nix-store/nix-store.cc b/src/nix-store/nix-store.cc index 062c68ff8..caa0248f1 100644 --- a/src/nix-store/nix-store.cc +++ b/src/nix-store/nix-store.cc @@ -807,6 +807,9 @@ static void opServe(Strings opFlags, Strings opArgs) out.flush(); unsigned int clientVersion = readInt(in); + WorkerProto::ReadConn rconn { .from = in }; + WorkerProto::WriteConn wconn { .to = out }; + auto getBuildSettings = [&]() { // FIXME: changing options here doesn't work if we're // building through the daemon. @@ -850,7 +853,7 @@ static void opServe(Strings opFlags, Strings opArgs) case ServeProto::Command::QueryValidPaths: { bool lock = readInt(in); bool substitute = readInt(in); - auto paths = WorkerProto::Serialise::read(*store, in); + auto paths = WorkerProto::Serialise::read(*store, rconn); if (lock && writeAllowed) for (auto & path : paths) store->addTempRoot(path); @@ -859,19 +862,19 @@ static void opServe(Strings opFlags, Strings opArgs) store->substitutePaths(paths); } - WorkerProto::write(*store, out, store->queryValidPaths(paths)); + WorkerProto::write(*store, wconn, store->queryValidPaths(paths)); break; } case ServeProto::Command::QueryPathInfos: { - auto paths = WorkerProto::Serialise::read(*store, in); + auto paths = WorkerProto::Serialise::read(*store, rconn); // !!! Maybe we want a queryPathInfos? for (auto & i : paths) { try { auto info = store->queryPathInfo(i); out << store->printStorePath(info->path) << (info->deriver ? store->printStorePath(*info->deriver) : ""); - WorkerProto::write(*store, out, info->references); + WorkerProto::write(*store, wconn, info->references); // !!! Maybe we want compression? out << info->narSize // downloadSize << info->narSize; @@ -899,7 +902,7 @@ static void opServe(Strings opFlags, Strings opArgs) case ServeProto::Command::ExportPaths: { readInt(in); // obsolete - store->exportPaths(WorkerProto::Serialise::read(*store, in), out); + store->exportPaths(WorkerProto::Serialise::read(*store, rconn), out); break; } @@ -945,7 +948,7 @@ static void opServe(Strings opFlags, Strings opArgs) DrvOutputs builtOutputs; for (auto & [output, realisation] : status.builtOutputs) builtOutputs.insert_or_assign(realisation.id, realisation); - WorkerProto::write(*store, out, builtOutputs); + WorkerProto::write(*store, wconn, builtOutputs); } break; @@ -954,9 +957,9 @@ static void opServe(Strings opFlags, Strings opArgs) case ServeProto::Command::QueryClosure: { bool includeOutputs = readInt(in); StorePathSet closure; - store->computeFSClosure(WorkerProto::Serialise::read(*store, in), + store->computeFSClosure(WorkerProto::Serialise::read(*store, rconn), closure, false, includeOutputs); - WorkerProto::write(*store, out, closure); + WorkerProto::write(*store, wconn, closure); break; } @@ -971,7 +974,7 @@ static void opServe(Strings opFlags, Strings opArgs) }; if (deriver != "") info.deriver = store->parseStorePath(deriver); - info.references = WorkerProto::Serialise::read(*store, in); + info.references = WorkerProto::Serialise::read(*store, rconn); in >> info.registrationTime >> info.narSize >> info.ultimate; info.sigs = readStrings(in); info.ca = ContentAddress::parseOpt(readString(in)); diff --git a/src/nix/daemon.cc b/src/nix/daemon.cc index 9fe9b3b1e..8e2bcf7e1 100644 --- a/src/nix/daemon.cc +++ b/src/nix/daemon.cc @@ -4,6 +4,7 @@ #include "shared.hh" #include "local-store.hh" #include "remote-store.hh" +#include "remote-store-connection.hh" #include "util.hh" #include "serialise.hh" #include "archive.hh" From 3859cf6b21dddd7e9de729dc42b2f52d8523c239 Mon Sep 17 00:00:00 2001 From: John Ericson Date: Mon, 19 Jun 2023 12:18:04 -0400 Subject: [PATCH 5/5] Remove unused `#include` from `local-derivation-goal.cc` These were never needed for this file, and date back to before this was split from `derivation-goal.cc`. --- src/libstore/build/local-derivation-goal.cc | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/libstore/build/local-derivation-goal.cc b/src/libstore/build/local-derivation-goal.cc index 6e06f6061..09c1fcf08 100644 --- a/src/libstore/build/local-derivation-goal.cc +++ b/src/libstore/build/local-derivation-goal.cc @@ -10,8 +10,6 @@ #include "archive.hh" #include "compression.hh" #include "daemon.hh" -#include "worker-protocol.hh" -#include "worker-protocol-impl.hh" #include "topo-sort.hh" #include "callback.hh" #include "json-utils.hh"