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

libstore: associate wire connection states with stores

Browse Source 
why pass the stores as a distinct argument every time?

Change-Id: If529a49541483e8a3d33eb2b3532d66b3bb9738d
This commit is contained in:
eldritch horrors
2025-06-06 18:09:46 +02:00
parent ce9acd5f97
commit 4ebf79bc19
20 changed files with 236 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/legacy/nix-store.cc
+11 -9
View File
@@ -858,9 +858,11 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
ServeProto::ReadConn rconn {
.from = in,
.store = *store,
.version = clientVersion,
};
ServeProto::WriteConn wconn {
.store = *store,
.version = clientVersion,
};
@@ -907,7 +909,7 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
case ServeProto::Command::QueryValidPaths: {
bool lock = readInt(in);
bool substitute = readInt(in);
auto paths = ServeProto::Serialise<StorePathSet>::read(*store, rconn);
auto paths = ServeProto::Serialise<StorePathSet>::read(rconn);
if (lock && writeAllowed)
for (auto & path : paths)
aio.blockOn(store->addTempRoot(path));
@@ -917,18 +919,18 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
}
auto valid = aio.blockOn(store->queryValidPaths(paths));
out << ServeProto::write(*store, wconn, valid);
out << ServeProto::write(wconn, valid);
break;
}
case ServeProto::Command::QueryPathInfos: {
auto paths = ServeProto::Serialise<StorePathSet>::read(*store, rconn);
auto paths = ServeProto::Serialise<StorePathSet>::read(rconn);
// !!! Maybe we want a queryPathInfos?
for (auto & i : paths) {
try {
auto info = aio.blockOn(store->queryPathInfo(i));
out << store->printStorePath(info->path);
out << ServeProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
out << ServeProto::write(wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
} catch (InvalidPath &) {
}
}
@@ -951,7 +953,7 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
case ServeProto::Command::ExportPaths: {
readInt(in); // obsolete
aio.blockOn(store->exportPaths(
ServeProto::Serialise<StorePathSet>::read(*store, rconn), out
ServeProto::Serialise<StorePathSet>::read(rconn), out
));
break;
}
@@ -990,7 +992,7 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
MonitorFdHup monitor(in.fd);
auto status = aio.blockOn(store->buildDerivation(drvPath, drv));
out << ServeProto::write(*store, wconn, status);
out << ServeProto::write(wconn, status);
break;
}
@@ -998,12 +1000,12 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
bool includeOutputs = readInt(in);
StorePathSet closure;
aio.blockOn(store->computeFSClosure(
ServeProto::Serialise<StorePathSet>::read(*store, rconn),
ServeProto::Serialise<StorePathSet>::read(rconn),
closure,
false,
includeOutputs
));
out << ServeProto::write(*store, wconn, closure);
out << ServeProto::write(wconn, closure);
break;
}
@@ -1018,7 +1020,7 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
};
if (deriver != "")
info.deriver = store->parseStorePath(deriver);
info.references = ServeProto::Serialise<StorePathSet>::read(*store, rconn);
info.references = ServeProto::Serialise<StorePathSet>::read(rconn);
in >> info.registrationTime >> info.narSize >> info.ultimate;
info.sigs = readStrings<StringSet>(in);
info.ca = ContentAddress::parseOpt(readString(in));
lix/libstore/build/derivation-goal.cc
+2 -2
View File
@@ -1108,7 +1108,7 @@ HookReply DerivationGoal::tryBuildHook()
/* Tell the hook all the inputs that have to be copied to the
remote system. */
hook->sink << CommonProto::write(worker.store, {}, inputPaths);
hook->sink << CommonProto::write({worker.store}, inputPaths);
/* Tell the hooks the missing outputs that have to be copied back
from the remote system. */
@@ -1119,7 +1119,7 @@ HookReply DerivationGoal::tryBuildHook()
if (buildMode != bmCheck && status.known && status.known->isValid()) continue;
missingOutputs.insert(outputName);
}
hook->sink << CommonProto::write(worker.store, {}, missingOutputs);
hook->sink << CommonProto::write({worker.store}, missingOutputs);
}
hook->sink = FdSink();
lix/libstore/common-protocol-impl.hh
+4 -4
View File
@@ -16,14 +16,14 @@ namespace nix {
/* protocol-agnostic templates */
#define COMMON_USE_LENGTH_PREFIX_SERIALISER(TEMPLATE, T) \
TEMPLATE T CommonProto::Serialise< T >::read(const Store & store, CommonProto::ReadConn conn) \
TEMPLATE T CommonProto::Serialise< T >::read(CommonProto::ReadConn conn) \
{ \
return LengthPrefixedProtoHelper<CommonProto, T >::read(store, conn); \
return LengthPrefixedProtoHelper<CommonProto, T >::read(conn); \
} \
/* NOLINTNEXTLINE(bugprone-macro-parentheses) */ \
TEMPLATE [[nodiscard]] WireFormatGenerator CommonProto::Serialise< T >::write(const Store & store, CommonProto::WriteConn conn, const T & t) \
TEMPLATE [[nodiscard]] WireFormatGenerator CommonProto::Serialise< T >::write(CommonProto::WriteConn conn, const T & t) \
{ \
return LengthPrefixedProtoHelper<CommonProto, T >::write(store, conn, t); \
return LengthPrefixedProtoHelper<CommonProto, T >::write(conn, t); \
}
COMMON_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
lix/libstore/common-protocol.cc
+18 -18
View File
@@ -10,40 +10,40 @@ namespace nix {
/* protocol-agnostic definitions */
std::string CommonProto::Serialise<std::string>::read(const Store & store, CommonProto::ReadConn conn)
std::string CommonProto::Serialise<std::string>::read(CommonProto::ReadConn conn)
{
return readString(conn.from);
}
WireFormatGenerator CommonProto::Serialise<std::string>::write(const Store & store, CommonProto::WriteConn conn, const std::string & str)
WireFormatGenerator CommonProto::Serialise<std::string>::write(CommonProto::WriteConn conn, const std::string & str)
{
co_yield str;
}
StorePath CommonProto::Serialise<StorePath>::read(const Store & store, CommonProto::ReadConn conn)
StorePath CommonProto::Serialise<StorePath>::read(CommonProto::ReadConn conn)
{
return store.parseStorePath(readString(conn.from));
return conn.store.parseStorePath(readString(conn.from));
}
WireFormatGenerator CommonProto::Serialise<StorePath>::write(const Store & store, CommonProto::WriteConn conn, const StorePath & storePath)
WireFormatGenerator CommonProto::Serialise<StorePath>::write(CommonProto::WriteConn conn, const StorePath & storePath)
{
co_yield store.printStorePath(storePath);
co_yield conn.store.printStorePath(storePath);
}
ContentAddress CommonProto::Serialise<ContentAddress>::read(const Store & store, CommonProto::ReadConn conn)
ContentAddress CommonProto::Serialise<ContentAddress>::read(CommonProto::ReadConn conn)
{
return ContentAddress::parse(readString(conn.from));
}
WireFormatGenerator CommonProto::Serialise<ContentAddress>::write(const Store & store, CommonProto::WriteConn conn, const ContentAddress & ca)
WireFormatGenerator CommonProto::Serialise<ContentAddress>::write(CommonProto::WriteConn conn, const ContentAddress & ca)
{
co_yield renderContentAddress(ca);
}
Realisation CommonProto::Serialise<Realisation>::read(const Store & store, CommonProto::ReadConn conn)
Realisation CommonProto::Serialise<Realisation>::read(CommonProto::ReadConn conn)
{
std::string rawInput = readString(conn.from);
return Realisation::fromJSON(
@@ -52,43 +52,43 @@ Realisation CommonProto::Serialise<Realisation>::read(const Store & store, Commo
);
}
WireFormatGenerator CommonProto::Serialise<Realisation>::write(const Store & store, CommonProto::WriteConn conn, const Realisation & realisation)
WireFormatGenerator CommonProto::Serialise<Realisation>::write(CommonProto::WriteConn conn, const Realisation & realisation)
{
co_yield realisation.toJSON().dump();
}
DrvOutput CommonProto::Serialise<DrvOutput>::read(const Store & store, CommonProto::ReadConn conn)
DrvOutput CommonProto::Serialise<DrvOutput>::read(CommonProto::ReadConn conn)
{
return DrvOutput::parse(readString(conn.from));
}
WireFormatGenerator CommonProto::Serialise<DrvOutput>::write(const Store & store, CommonProto::WriteConn conn, const DrvOutput & drvOutput)
WireFormatGenerator CommonProto::Serialise<DrvOutput>::write(CommonProto::WriteConn conn, const DrvOutput & drvOutput)
{
co_yield drvOutput.to_string();
}
std::optional<StorePath> CommonProto::Serialise<std::optional<StorePath>>::read(const Store & store, CommonProto::ReadConn conn)
std::optional<StorePath> CommonProto::Serialise<std::optional<StorePath>>::read(CommonProto::ReadConn conn)
{
auto s = readString(conn.from);
return s == "" ? std::optional<StorePath> {} : store.parseStorePath(s);
return s == "" ? std::optional<StorePath> {} : conn.store.parseStorePath(s);
}
WireFormatGenerator CommonProto::Serialise<std::optional<StorePath>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<StorePath> & storePathOpt)
WireFormatGenerator CommonProto::Serialise<std::optional<StorePath>>::write(CommonProto::WriteConn conn, const std::optional<StorePath> & storePathOpt)
{
return [](std::string s) -> WireFormatGenerator {
co_yield s;
}(storePathOpt ? store.printStorePath(*storePathOpt) : "");
}(storePathOpt ? conn.store.printStorePath(*storePathOpt) : "");
}
std::optional<ContentAddress> CommonProto::Serialise<std::optional<ContentAddress>>::read(const Store & store, CommonProto::ReadConn conn)
std::optional<ContentAddress> CommonProto::Serialise<std::optional<ContentAddress>>::read(CommonProto::ReadConn conn)
{
return ContentAddress::parseOpt(readString(conn.from));
}
WireFormatGenerator CommonProto::Serialise<std::optional<ContentAddress>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<ContentAddress> & caOpt)
WireFormatGenerator CommonProto::Serialise<std::optional<ContentAddress>>::write(CommonProto::WriteConn conn, const std::optional<ContentAddress> & caOpt)
{
return [](std::string s) -> WireFormatGenerator {
co_yield s;
lix/libstore/common-protocol.hh
+6 -4
View File
@@ -30,6 +30,7 @@ struct CommonProto
*/
struct ReadConn {
Source & from;
const Store & store;
};
/**
@@ -37,6 +38,7 @@ struct CommonProto
* canonical serializers below.
*/
struct WriteConn {
const Store & store;
};
template<typename T>
@@ -48,17 +50,17 @@ struct CommonProto
*/
template<typename T>
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
static WireFormatGenerator write(WriteConn conn, const T & t)
{
return CommonProto::Serialise<T>::write(store, conn, t);
return CommonProto::Serialise<T>::write(conn, t);
}
};
#define DECLARE_COMMON_SERIALISER(T) \
struct CommonProto::Serialise< T > \
{ \
static T read(const Store & store, CommonProto::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, CommonProto::WriteConn conn, const T & str); \
static T read(CommonProto::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(CommonProto::WriteConn conn, const T & str); \
}
template<>
lix/libstore/daemon.cc
+28 -28
View File
@@ -238,8 +238,8 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
TrustedFlag trusted, WorkerProto::Version clientVersion,
Source & from, BufferedSink & to, WorkerProto::Op op)
{
WorkerProto::ReadConn rconn{from, clientVersion};
WorkerProto::WriteConn wconn{clientVersion};
WorkerProto::ReadConn rconn{from, *store, clientVersion};
WorkerProto::WriteConn wconn{*store, clientVersion};
switch (op) {
@@ -253,7 +253,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
case WorkerProto::Op::QueryValidPaths: {
auto paths = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
auto paths = WorkerProto::Serialise<StorePathSet>::read(rconn);
SubstituteFlag substitute = readInt(from) ? Substitute : NoSubstitute;
@@ -263,16 +263,16 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
auto res = aio.blockOn(store->queryValidPaths(paths, substitute));
logger->stopWork();
to << WorkerProto::write(*store, wconn, res);
to << WorkerProto::write(wconn, res);
break;
}
case WorkerProto::Op::QuerySubstitutablePaths: {
auto paths = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
auto paths = WorkerProto::Serialise<StorePathSet>::read(rconn);
logger->startWork();
auto res = aio.blockOn(store->querySubstitutablePaths(paths));
logger->stopWork();
to << WorkerProto::write(*store, wconn, res);
to << WorkerProto::write(wconn, res);
break;
}
@@ -334,7 +334,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
#pragma GCC diagnostic pop
logger->stopWork();
to << WorkerProto::write(*store, wconn, paths);
to << WorkerProto::write(wconn, paths);
break;
}
@@ -347,7 +347,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
logger->startWork();
auto outputs = aio.blockOn(store->queryDerivationOutputMap(path));
logger->stopWork();
to << WorkerProto::write(*store, wconn, outputs);
to << WorkerProto::write(wconn, outputs);
break;
}
@@ -363,7 +363,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
case WorkerProto::Op::AddToStore: {
auto name = readString(from);
auto camStr = readString(from);
auto refs = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
auto refs = WorkerProto::Serialise<StorePathSet>::read(rconn);
bool repairBool;
from >> repairBool;
auto repair = RepairFlag{repairBool};
@@ -396,7 +396,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}();
logger->stopWork();
to << WorkerProto::Serialise<ValidPathInfo>::write(*store, wconn, *pathInfo);
to << WorkerProto::Serialise<ValidPathInfo>::write(wconn, *pathInfo);
break;
}
@@ -412,7 +412,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
auto expected = readNum<uint64_t>(source);
for (uint64_t i = 0; i < expected; ++i) {
auto info = WorkerProto::Serialise<ValidPathInfo>::read(
*store, WorkerProto::ReadConn{source, clientVersion}
WorkerProto::ReadConn{source, *store, clientVersion}
);
info.ultimate = false; // duplicated in RemoteStore::addMultipleToStore
AsyncSourceInputStream stream{source};
@@ -430,7 +430,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
case WorkerProto::Op::BuildPaths: {
auto drvs = WorkerProto::Serialise<DerivedPaths>::read(*store, rconn);
auto drvs = WorkerProto::Serialise<DerivedPaths>::read(rconn);
BuildMode mode = buildModeFromInteger(readInt(from));
/* Repairing is not atomic, so disallowed for "untrusted"
@@ -452,7 +452,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
case WorkerProto::Op::BuildPathsWithResults: {
auto drvs = WorkerProto::Serialise<DerivedPaths>::read(*store, rconn);
auto drvs = WorkerProto::Serialise<DerivedPaths>::read(rconn);
BuildMode mode = bmNormal;
mode = buildModeFromInteger(readInt(from));
@@ -467,7 +467,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
auto results = aio.blockOn(store->buildPathsWithResults(drvs, mode));
logger->stopWork();
to << WorkerProto::write(*store, wconn, results);
to << WorkerProto::write(wconn, results);
break;
}
@@ -545,7 +545,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
auto res = aio.blockOn(store->buildDerivation(drvPath, drv, buildMode));
logger->stopWork();
to << WorkerProto::write(*store, wconn, res);
to << WorkerProto::write(wconn, res);
break;
}
@@ -606,7 +606,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
case WorkerProto::Op::CollectGarbage: {
GCOptions options;
options.action = (GCOptions::GCAction) readInt(from);
options.pathsToDelete = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
options.pathsToDelete = WorkerProto::Serialise<StorePathSet>::read(rconn);
from >> options.ignoreLiveness >> options.maxFreed;
// obsolete fields
readInt(from);
@@ -669,7 +669,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
else {
to << 1
<< (i->second.deriver ? store->printStorePath(*i->second.deriver) : "");
to << WorkerProto::write(*store, wconn, i->second.references);
to << WorkerProto::write(wconn, i->second.references);
to << i->second.downloadSize
<< i->second.narSize;
}
@@ -678,11 +678,11 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
case WorkerProto::Op::QuerySubstitutablePathInfos: {
SubstitutablePathInfos infos;
StorePathCAMap pathsMap = WorkerProto::Serialise<StorePathCAMap>::read(*store, rconn);
StorePathCAMap pathsMap = WorkerProto::Serialise<StorePathCAMap>::read(rconn);
logger->startWork();
aio.blockOn(store->querySubstitutablePathInfos(pathsMap, infos));
logger->stopWork();
to << WorkerProto::write(*store, wconn, infos);
to << WorkerProto::write(wconn, infos);
break;
}
@@ -690,7 +690,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
logger->startWork();
auto paths = aio.blockOn(store->queryAllValidPaths());
logger->stopWork();
to << WorkerProto::write(*store, wconn, paths);
to << WorkerProto::write(wconn, paths);
break;
}
@@ -707,7 +707,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
logger->stopWork();
if (info) {
to << 1;
to << WorkerProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
to << WorkerProto::write(wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
} else {
to << 0;
}
@@ -759,7 +759,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
ValidPathInfo info { path, narHash };
if (deriver != "")
info.deriver = store->parseStorePath(deriver);
info.references = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
info.references = WorkerProto::Serialise<StorePathSet>::read(rconn);
from >> info.registrationTime >> info.narSize >> info.ultimate;
info.sigs = readStrings<StringSet>(from);
info.ca = ContentAddress::parseOpt(readString(from));
@@ -782,7 +782,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
case WorkerProto::Op::QueryMissing: {
auto targets = WorkerProto::Serialise<DerivedPaths>::read(*store, rconn);
auto targets = WorkerProto::Serialise<DerivedPaths>::read(rconn);
logger->startWork();
StorePathSet willBuild, willSubstitute, unknown;
uint64_t downloadSize, narSize;
@@ -790,9 +790,9 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
store->queryMissing(targets, willBuild, willSubstitute, unknown, downloadSize, narSize)
);
logger->stopWork();
to << WorkerProto::write(*store, wconn, willBuild);
to << WorkerProto::write(*store, wconn, willSubstitute);
to << WorkerProto::write(*store, wconn, unknown);
to << WorkerProto::write(wconn, willBuild);
to << WorkerProto::write(wconn, willSubstitute);
to << WorkerProto::write(wconn, unknown);
to << downloadSize << narSize;
break;
}
@@ -873,8 +873,8 @@ void processConnection(
auto temp = trusted
? aio.blockOn(store->isTrustedClient())
: std::optional { NotTrusted };
WorkerProto::WriteConn wconn {clientVersion};
to << WorkerProto::write(*store, wconn, temp);
WorkerProto::WriteConn wconn {*store, clientVersion};
to << WorkerProto::write(wconn, temp);
/* Send startup error messages to the client. */
tunnelLogger->startWork();
lix/libstore/derivations.cc
+3 -5
View File
@@ -676,8 +676,8 @@ Source & readDerivation(Source & in, const Store & store, BasicDerivation & drv,
drv.outputs.emplace(std::move(name), std::move(output));
}
drv.inputSrcs = CommonProto::Serialise<StorePathSet>::read(store,
CommonProto::ReadConn { .from = in });
drv.inputSrcs = CommonProto::Serialise<StorePathSet>::read(
CommonProto::ReadConn { .from = in, .store = store });
in >> drv.platform >> drv.builder;
drv.args = readStrings<Strings>(in);
@@ -710,9 +710,7 @@ void writeDerivation(Sink & out, const Store & store, const BasicDerivation & dr
},
}, i.second.raw);
}
out << CommonProto::write(store,
CommonProto::WriteConn {},
drv.inputSrcs);
out << CommonProto::write(CommonProto::WriteConn{store}, drv.inputSrcs);
out << drv.platform << drv.builder << drv.args;
out << drv.env.size();
for (auto & i : drv.env)
lix/libstore/export-import.cc
+4 -5
View File
@@ -45,9 +45,7 @@ try {
teeSink
<< exportMagic
<< printStorePath(path);
teeSink << CommonProto::write(*this,
CommonProto::WriteConn {},
info->references);
teeSink << CommonProto::write(CommonProto::WriteConn{*this}, info->references);
teeSink
<< (info->deriver ? printStorePath(*info->deriver) : "")
<< 0;
@@ -76,8 +74,9 @@ try {
//Activity act(*logger, lvlInfo, "importing path '%s'", info.path);
auto references = CommonProto::Serialise<StorePathSet>::read(*this,
CommonProto::ReadConn { .from = source });
auto references = CommonProto::Serialise<StorePathSet>::read(
CommonProto::ReadConn{.from = source, .store = *this}
);
auto deriver = readString(source);
auto narHash = hashString(HashType::SHA256, saved.s);
lix/libstore/legacy-ssh-store.cc
+12 -8
View File
@@ -60,6 +60,7 @@ struct LegacySSHStore final : public Store
FdSink to;
FdSource from;
ServeProto::Version remoteVersion;
Store * store = nullptr;
bool good = true;
/**
@@ -74,6 +75,7 @@ struct LegacySSHStore final : public Store
{
return ServeProto::ReadConn {
.from = from,
.store = *store,
.version = remoteVersion,
};
}
@@ -89,6 +91,7 @@ struct LegacySSHStore final : public Store
operator ServeProto::WriteConn ()
{
return ServeProto::WriteConn {
.store = *store,
.version = remoteVersion,
};
}
@@ -134,6 +137,7 @@ struct LegacySSHStore final : public Store
);
conn->to = FdSink(conn->sshConn->in.get());
conn->from = FdSource(conn->sshConn->out.get());
conn->store = this;
try {
conn->to << SERVE_MAGIC_1 << SERVE_PROTOCOL_VERSION;
@@ -177,7 +181,7 @@ struct LegacySSHStore final : public Store
assert(path == path2);
auto info = std::make_shared<ValidPathInfo>(
path,
ServeProto::Serialise<UnkeyedValidPathInfo>::read(*this, *conn));
ServeProto::Serialise<UnkeyedValidPathInfo>::read(*conn));
if (info->narHash == Hash::dummy)
throw Error("NAR hash is now mandatory");
@@ -204,7 +208,7 @@ struct LegacySSHStore final : public Store
<< printStorePath(info.path)
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< info.narHash.to_string(Base::Base16, false);
conn->to << ServeProto::write(*this, *conn, info.references);
conn->to << ServeProto::write(*conn, info.references);
conn->to
<< info.registrationTime
<< info.narSize
@@ -233,7 +237,7 @@ struct LegacySSHStore final : public Store
conn->to
<< exportMagic
<< printStorePath(info.path);
conn->to << ServeProto::write(*this, *conn, info.references);
conn->to << ServeProto::write(*conn, info.references);
conn->to
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< 0
@@ -328,7 +332,7 @@ public:
conn->to.flush();
co_return ServeProto::Serialise<BuildResult>::read(*this, *conn);
co_return ServeProto::Serialise<BuildResult>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
@@ -412,10 +416,10 @@ public:
conn->to
<< ServeProto::Command::QueryClosure
<< includeOutputs;
conn->to << ServeProto::write(*this, *conn, paths);
conn->to << ServeProto::write(*conn, paths);
conn->to.flush();
for (auto & i : ServeProto::Serialise<StorePathSet>::read(*this, *conn))
for (auto & i : ServeProto::Serialise<StorePathSet>::read(*conn))
out.insert(i);
co_return result::success();
} catch (...) {
@@ -431,10 +435,10 @@ public:
<< ServeProto::Command::QueryValidPaths
<< false // lock
<< maybeSubstitute;
conn->to << ServeProto::write(*this, *conn, paths);
conn->to << ServeProto::write(*conn, paths);
conn->to.flush();
co_return ServeProto::Serialise<StorePathSet>::read(*this, *conn);
co_return ServeProto::Serialise<StorePathSet>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
lix/libstore/length-prefixed-protocol-helper.hh
+23 -24
View File
@@ -45,8 +45,8 @@ struct LengthPrefixedProtoHelper;
#define LENGTH_PREFIXED_PROTO_HELPER(Inner, T) \
struct LengthPrefixedProtoHelper< Inner, T > \
{ \
static T read(const Store & store, typename Inner::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, typename Inner::WriteConn conn, const T & str); \
static T read(typename Inner::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(typename Inner::WriteConn conn, const T & str); \
private: \
template<typename U> using S = typename Inner::template Serialise<U>; \
}
@@ -67,13 +67,12 @@ LENGTH_PREFIXED_PROTO_HELPER(Inner, DONT_SUBSTITUTE_KV_TYPE);
template<class Inner, typename T>
std::vector<T>
LengthPrefixedProtoHelper<Inner, std::vector<T>>::read(
const Store & store, typename Inner::ReadConn conn)
LengthPrefixedProtoHelper<Inner, std::vector<T>>::read(typename Inner::ReadConn conn)
{
std::vector<T> resSet;
auto size = readNum<size_t>(conn.from);
while (size--) {
resSet.push_back(S<T>::read(store, conn));
resSet.push_back(S<T>::read(conn));
}
return resSet;
}
@@ -81,23 +80,23 @@ LengthPrefixedProtoHelper<Inner, std::vector<T>>::read(
template<class Inner, typename T>
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::vector<T>>::write(
const Store & store, typename Inner::WriteConn conn, const std::vector<T> & resSet)
typename Inner::WriteConn conn, const std::vector<T> & resSet)
{
co_yield resSet.size();
for (auto & key : resSet) {
co_yield S<T>::write(store, conn, key);
co_yield S<T>::write(conn, key);
}
}
template<class Inner, typename T>
std::set<T>
LengthPrefixedProtoHelper<Inner, std::set<T>>::read(
const Store & store, typename Inner::ReadConn conn)
typename Inner::ReadConn conn)
{
std::set<T> resSet;
auto size = readNum<size_t>(conn.from);
while (size--) {
resSet.insert(S<T>::read(store, conn));
resSet.insert(S<T>::read(conn));
}
return resSet;
}
@@ -105,24 +104,24 @@ LengthPrefixedProtoHelper<Inner, std::set<T>>::read(
template<class Inner, typename T>
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::set<T>>::write(
const Store & store, typename Inner::WriteConn conn, const std::set<T> & resSet)
typename Inner::WriteConn conn, const std::set<T> & resSet)
{
co_yield resSet.size();
for (auto & key : resSet) {
co_yield S<T>::write(store, conn, key);
co_yield S<T>::write(conn, key);
}
}
template<class Inner, typename K, typename V>
std::map<K, V>
LengthPrefixedProtoHelper<Inner, std::map<K, V>>::read(
const Store & store, typename Inner::ReadConn conn)
typename Inner::ReadConn conn)
{
std::map<K, V> resMap;
auto size = readNum<size_t>(conn.from);
while (size--) {
auto k = S<K>::read(store, conn);
auto v = S<V>::read(store, conn);
auto k = S<K>::read(conn);
auto v = S<V>::read(conn);
resMap.insert_or_assign(std::move(k), std::move(v));
}
return resMap;
@@ -131,41 +130,41 @@ LengthPrefixedProtoHelper<Inner, std::map<K, V>>::read(
template<class Inner, typename K, typename V>
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::map<K, V>>::write(
const Store & store, typename Inner::WriteConn conn, const std::map<K, V> & resMap)
typename Inner::WriteConn conn, const std::map<K, V> & resMap)
{
co_yield resMap.size();
for (auto & i : resMap) {
co_yield S<K>::write(store, conn, i.first);
co_yield S<V>::write(store, conn, i.second);
co_yield S<K>::write(conn, i.first);
co_yield S<V>::write(conn, i.second);
}
}
template<class Inner, typename... Ts>
std::tuple<Ts...>
LengthPrefixedProtoHelper<Inner, std::tuple<Ts...>>::read(
const Store & store, typename Inner::ReadConn conn)
typename Inner::ReadConn conn)
{
return std::tuple<Ts...> {
S<Ts>::read(store, conn)...,
S<Ts>::read(conn)...,
};
}
template<class Inner, typename... Ts>
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::tuple<Ts...>>::write(
const Store & store, typename Inner::WriteConn conn, const std::tuple<Ts...> & res)
typename Inner::WriteConn conn, const std::tuple<Ts...> & res)
{
auto fullArgs = std::apply(
[&](auto &... rest) {
return std::tuple<const Store &, typename Inner::WriteConn &, const Ts &...>(
std::cref(store), conn, rest...
return std::tuple<typename Inner::WriteConn &, const Ts &...>(
conn, rest...
);
},
res
);
return std::apply(
[]<typename... Us>(auto & store, auto conn, const Us &... args) -> WireFormatGenerator {
(co_yield S<Us>::write(store, conn, args), ...);
[]<typename... Us>(auto conn, const Us &... args) -> WireFormatGenerator {
(co_yield S<Us>::write(conn, args), ...);
},
fullArgs
);
lix/libstore/remote-store-connection.hh
+7 -2
View File
@@ -26,6 +26,11 @@ struct RemoteStore::Connection
*/
FdSource from;
/**
* The store this connection belongs to.
*/
Store * store;
/**
* The worker protocol version of the connected daemon. This may be newer
* than this Lix supports.
@@ -68,7 +73,7 @@ struct RemoteStore::Connection
*/
operator WorkerProto::ReadConn ()
{
return WorkerProto::ReadConn {from, daemonVersion};
return WorkerProto::ReadConn{from, *store, daemonVersion};
}
/**
@@ -81,7 +86,7 @@ struct RemoteStore::Connection
*/
operator WorkerProto::WriteConn ()
{
return WorkerProto::WriteConn {daemonVersion};
return WorkerProto::WriteConn{*store, daemonVersion};
}
virtual ~Connection();
lix/libstore/remote-store.cc
+27 -26
View File
@@ -75,6 +75,7 @@ void RemoteStore::initConnection(Connection & conn)
{
/* Send the magic greeting, check for the reply. */
try {
conn.store = this;
conn.from.specialEndOfFileError = "Nix daemon disconnected unexpectedly (maybe it crashed?)";
conn.to << WORKER_MAGIC_1;
conn.to.flush();
@@ -97,7 +98,7 @@ void RemoteStore::initConnection(Connection & conn)
conn.to.flush();
conn.daemonNixVersion = readString(conn.from);
conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(*this, conn);
conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(conn);
auto ex = conn.processStderr();
if (ex) std::rethrow_exception(ex);
@@ -202,10 +203,10 @@ RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSub
try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryValidPaths;
conn->to << WorkerProto::write(*this, *conn, paths);
conn->to << WorkerProto::write(*conn, paths);
conn->to << maybeSubstitute;
conn.processStderr();
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
co_return WorkerProto::Serialise<StorePathSet>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
@@ -216,7 +217,7 @@ try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryAllValidPaths;
conn.processStderr();
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
co_return WorkerProto::Serialise<StorePathSet>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
@@ -226,9 +227,9 @@ kj::Promise<Result<StorePathSet>> RemoteStore::querySubstitutablePaths(const Sto
try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QuerySubstitutablePaths;
conn->to << WorkerProto::write(*this, *conn, paths);
conn->to << WorkerProto::write(*conn, paths);
conn.processStderr();
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
co_return WorkerProto::Serialise<StorePathSet>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
@@ -242,9 +243,9 @@ try {
conn->to << WorkerProto::Op::QuerySubstitutablePathInfos;
conn->to << WorkerProto::write(*this, *conn, pathsMap);
conn->to << WorkerProto::write(*conn, pathsMap);
conn.processStderr();
infos = WorkerProto::Serialise<SubstitutablePathInfos>::read(*this, *conn);
infos = WorkerProto::Serialise<SubstitutablePathInfos>::read(*conn);
co_return result::success();
} catch (...) {
co_return result::current_exception();
@@ -270,7 +271,7 @@ try {
co_return std::make_shared<ValidPathInfo>(
StorePath{path},
WorkerProto::Serialise<UnkeyedValidPathInfo>::read(*this, *conn));
WorkerProto::Serialise<UnkeyedValidPathInfo>::read(*conn));
} catch (...) {
co_return result::current_exception();
}
@@ -282,7 +283,7 @@ try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryReferrers << printStorePath(path);
conn.processStderr();
for (auto & i : WorkerProto::Serialise<StorePathSet>::read(*this, *conn))
for (auto & i : WorkerProto::Serialise<StorePathSet>::read(*conn))
referrers.insert(i);
co_return result::success();
} catch (...) {
@@ -295,7 +296,7 @@ try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryValidDerivers << printStorePath(path);
conn.processStderr();
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
co_return WorkerProto::Serialise<StorePathSet>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
@@ -309,7 +310,7 @@ try {
conn->to << WorkerProto::Op::QueryDerivationOutputMap << printStorePath(path);
conn.processStderr();
auto tmp = WorkerProto::Serialise<std::map<std::string, std::optional<StorePath>>>::read(
*this, *conn
*conn
);
std::map<std::string, StorePath> result;
for (auto & [name, outPath] : tmp) {
@@ -365,7 +366,7 @@ try {
<< WorkerProto::Op::AddToStore
<< name
<< caMethod.render(hashType);
conn->to << WorkerProto::write(*this, *conn, references);
conn->to << WorkerProto::write(*conn, references);
conn->to << repair;
// The dump source may invoke the store, so we need to make some room.
@@ -378,7 +379,7 @@ try {
}
co_return make_ref<ValidPathInfo>(
WorkerProto::Serialise<ValidPathInfo>::read(*this, *conn));
WorkerProto::Serialise<ValidPathInfo>::read(*conn));
} catch (...) {
co_return result::current_exception();
}
@@ -412,7 +413,7 @@ try {
<< printStorePath(info.path)
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< info.narHash.to_string(Base::Base16, false);
conn->to << WorkerProto::write(*this, *conn, info.references);
conn->to << WorkerProto::write(*conn, info.references);
conn->to << info.registrationTime << info.narSize
<< info.ultimate << info.sigs << renderContentAddress(info.ca)
<< repair << !checkSigs;
@@ -445,8 +446,8 @@ try {
try {
sink << pathsToCopy.size();
for (auto & [pathInfo, pathSource] : pathsToCopy) {
sink << WorkerProto::Serialise<ValidPathInfo>::write(*this,
WorkerProto::WriteConn {remoteVersion},
sink << WorkerProto::Serialise<ValidPathInfo>::write(
WorkerProto::WriteConn {*this, remoteVersion},
pathInfo);
TRY_AWAIT(TRY_AWAIT(pathSource())->drainInto(sink));
}
@@ -506,7 +507,7 @@ try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::BuildPaths;
conn->to << WorkerProto::write(*this, *conn, drvPaths);
conn->to << WorkerProto::write(*conn, drvPaths);
conn->to << buildMode;
conn.processStderr();
readInt(conn->from);
@@ -525,10 +526,10 @@ try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::BuildPathsWithResults;
conn->to << WorkerProto::write(*this, *conn, paths);
conn->to << WorkerProto::write(*conn, paths);
conn->to << buildMode;
conn.processStderr();
co_return WorkerProto::Serialise<std::vector<KeyedBuildResult>>::read(*this, *conn);
co_return WorkerProto::Serialise<std::vector<KeyedBuildResult>>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
@@ -542,7 +543,7 @@ try {
writeDerivation(conn->to, *this, drv);
conn->to << buildMode;
conn.processStderr();
co_return WorkerProto::Serialise<BuildResult>::read(*this, *conn);
co_return WorkerProto::Serialise<BuildResult>::read(*conn);
} catch (...) {
co_return result::current_exception();
}
@@ -597,7 +598,7 @@ try {
conn->to
<< WorkerProto::Op::CollectGarbage << options.action;
conn->to << WorkerProto::write(*this, *conn, options.pathsToDelete);
conn->to << WorkerProto::write(*conn, options.pathsToDelete);
conn->to << options.ignoreLiveness
<< options.maxFreed
/* removed options */
@@ -661,11 +662,11 @@ kj::Promise<Result<void>> RemoteStore::queryMissing(const std::vector<DerivedPat
try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryMissing;
conn->to << WorkerProto::write(*this, *conn, targets);
conn->to << WorkerProto::write(*conn, targets);
conn.processStderr();
willBuild = WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
willSubstitute = WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
unknown = WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
willBuild = WorkerProto::Serialise<StorePathSet>::read(*conn);
willSubstitute = WorkerProto::Serialise<StorePathSet>::read(*conn);
unknown = WorkerProto::Serialise<StorePathSet>::read(*conn);
conn->from >> downloadSize >> narSize;
co_return result::success();
} catch (...) {
lix/libstore/serve-protocol-impl.hh
+10 -11
View File
@@ -16,14 +16,14 @@ namespace nix {
/* protocol-agnostic templates */
#define SERVE_USE_LENGTH_PREFIX_SERIALISER(TEMPLATE, T) \
TEMPLATE T ServeProto::Serialise< T >::read(const Store & store, ServeProto::ReadConn conn) \
TEMPLATE T ServeProto::Serialise< T >::read(ServeProto::ReadConn conn) \
{ \
return LengthPrefixedProtoHelper<ServeProto, T >::read(store, conn); \
return LengthPrefixedProtoHelper<ServeProto, T >::read(conn); \
} \
/* NOLINTNEXTLINE(bugprone-macro-parentheses) */ \
TEMPLATE [[nodiscard]] WireFormatGenerator ServeProto::Serialise< T >::write(const Store & store, ServeProto::WriteConn conn, const T & t) \
TEMPLATE [[nodiscard]] WireFormatGenerator ServeProto::Serialise< T >::write(ServeProto::WriteConn conn, const T & t) \
{ \
return LengthPrefixedProtoHelper<ServeProto, T >::write(store, conn, t); \
return LengthPrefixedProtoHelper<ServeProto, T >::write(conn, t); \
}
SERVE_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
@@ -42,17 +42,16 @@ SERVE_USE_LENGTH_PREFIX_SERIALISER(
template<typename T>
struct ServeProto::Serialise
{
static T read(const Store & store, ServeProto::ReadConn conn)
static T read(ServeProto::ReadConn conn)
{
return CommonProto::Serialise<T>::read(store,
CommonProto::ReadConn { .from = conn.from });
return CommonProto::Serialise<T>::read(
CommonProto::ReadConn{.from = conn.from, .store = conn.store}
);
}
[[nodiscard]]
static WireFormatGenerator write(const Store & store, ServeProto::WriteConn conn, const T & t)
static WireFormatGenerator write(ServeProto::WriteConn conn, const T & t)
{
return CommonProto::Serialise<T>::write(store,
CommonProto::WriteConn {},
t);
return CommonProto::Serialise<T>::write(CommonProto::WriteConn{conn.store}, t);
}
};
lix/libstore/serve-protocol.cc
+10 -10
View File
@@ -10,7 +10,7 @@ namespace nix {
/* protocol-specific definitions */
BuildResult ServeProto::Serialise<BuildResult>::read(const Store & store, ServeProto::ReadConn conn)
BuildResult ServeProto::Serialise<BuildResult>::read(ServeProto::ReadConn conn)
{
BuildResult status;
status.status = (BuildResult::Status) readInt(conn.from);
@@ -23,7 +23,7 @@ BuildResult ServeProto::Serialise<BuildResult>::read(const Store & store, ServeP
>> status.startTime
>> status.stopTime;
if (GET_PROTOCOL_MINOR(conn.version) >= 6) {
auto builtOutputs = ServeProto::Serialise<DrvOutputs>::read(store, conn);
auto builtOutputs = ServeProto::Serialise<DrvOutputs>::read(conn);
for (auto && [output, realisation] : builtOutputs)
status.builtOutputs.insert_or_assign(
std::move(output.outputName),
@@ -32,7 +32,7 @@ BuildResult ServeProto::Serialise<BuildResult>::read(const Store & store, ServeP
return status;
}
WireFormatGenerator ServeProto::Serialise<BuildResult>::write(const Store & store, ServeProto::WriteConn conn, const BuildResult & status)
WireFormatGenerator ServeProto::Serialise<BuildResult>::write(ServeProto::WriteConn conn, const BuildResult & status)
{
co_yield status.status;
co_yield status.errorMsg;
@@ -47,12 +47,12 @@ WireFormatGenerator ServeProto::Serialise<BuildResult>::write(const Store & stor
DrvOutputs builtOutputs;
for (auto & [output, realisation] : status.builtOutputs)
builtOutputs.insert_or_assign(realisation.id, realisation);
co_yield ServeProto::write(store, conn, builtOutputs);
co_yield ServeProto::write(conn, builtOutputs);
}
}
UnkeyedValidPathInfo ServeProto::Serialise<UnkeyedValidPathInfo>::read(const Store & store, ReadConn conn)
UnkeyedValidPathInfo ServeProto::Serialise<UnkeyedValidPathInfo>::read(ReadConn conn)
{
/* Hash should be set below unless very old `nix-store --serve`.
Caller should assert that it did set it. */
@@ -60,8 +60,8 @@ UnkeyedValidPathInfo ServeProto::Serialise<UnkeyedValidPathInfo>::read(const Sto
auto deriver = readString(conn.from);
if (deriver != "")
info.deriver = store.parseStorePath(deriver);
info.references = ServeProto::Serialise<StorePathSet>::read(store, conn);
info.deriver = conn.store.parseStorePath(deriver);
info.references = ServeProto::Serialise<StorePathSet>::read(conn);
readLongLong(conn.from); // download size, unused
info.narSize = readLongLong(conn.from);
@@ -77,11 +77,11 @@ UnkeyedValidPathInfo ServeProto::Serialise<UnkeyedValidPathInfo>::read(const Sto
return info;
}
WireFormatGenerator ServeProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & info)
WireFormatGenerator ServeProto::Serialise<UnkeyedValidPathInfo>::write(WriteConn conn, const UnkeyedValidPathInfo & info)
{
co_yield (info.deriver ? store.printStorePath(*info.deriver) : "");
co_yield (info.deriver ? conn.store.printStorePath(*info.deriver) : "");
co_yield ServeProto::write(store, conn, info.references);
co_yield ServeProto::write(conn, info.references);
// !!! Maybe we want compression?
co_yield info.narSize; // downloadSize, lie a little
co_yield info.narSize;
lix/libstore/serve-protocol.hh
+8 -6
View File
@@ -52,6 +52,7 @@ struct ServeProto
*/
struct ReadConn {
Source & from;
const Store & store;
Version version;
};
@@ -60,6 +61,7 @@ struct ServeProto
* canonical serializers below.
*/
struct WriteConn {
const Store & store;
Version version;
};
@@ -77,8 +79,8 @@ struct ServeProto
// See `worker-protocol.hh` for a longer explanation.
#if 0
{
static T read(const Store & store, ReadConn conn);
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t);
static T read(ReadConn conn);
static WireFormatGenerator write(WriteConn conn, const T & t);
};
#endif
@@ -88,9 +90,9 @@ struct ServeProto
*/
template<typename T>
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
static WireFormatGenerator write(WriteConn conn, const T & t)
{
return ServeProto::Serialise<T>::write(store, conn, t);
return ServeProto::Serialise<T>::write(conn, t);
}
};
@@ -141,8 +143,8 @@ inline std::ostream & operator << (std::ostream & s, ServeProto::Command op)
#define DECLARE_SERVE_SERIALISER(T) \
struct ServeProto::Serialise< T > \
{ \
static T read(const Store & store, ServeProto::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, ServeProto::WriteConn conn, const T & t); \
static T read(ServeProto::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(ServeProto::WriteConn conn, const T & t); \
};
template<>
lix/libstore/worker-protocol-impl.hh
+10 -11
View File
@@ -16,14 +16,14 @@ namespace nix {
/* protocol-agnostic templates */
#define WORKER_USE_LENGTH_PREFIX_SERIALISER(TEMPLATE, T) \
TEMPLATE T WorkerProto::Serialise< T >::read(const Store & store, WorkerProto::ReadConn conn) \
TEMPLATE T WorkerProto::Serialise< T >::read(WorkerProto::ReadConn conn) \
{ \
return LengthPrefixedProtoHelper<WorkerProto, T >::read(store, conn); \
return LengthPrefixedProtoHelper<WorkerProto, T >::read(conn); \
} \
/* NOLINTNEXTLINE(bugprone-macro-parentheses) */ \
TEMPLATE [[nodiscard]] WireFormatGenerator WorkerProto::Serialise< T >::write(const Store & store, WorkerProto::WriteConn conn, const T & t) \
TEMPLATE [[nodiscard]] WireFormatGenerator WorkerProto::Serialise< T >::write(WorkerProto::WriteConn conn, const T & t) \
{ \
return LengthPrefixedProtoHelper<WorkerProto, T >::write(store, conn, t); \
return LengthPrefixedProtoHelper<WorkerProto, T >::write(conn, t); \
}
WORKER_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
@@ -42,17 +42,16 @@ WORKER_USE_LENGTH_PREFIX_SERIALISER(
template<typename T>
struct WorkerProto::Serialise
{
static T read(const Store & store, WorkerProto::ReadConn conn)
static T read(WorkerProto::ReadConn conn)
{
return CommonProto::Serialise<T>::read(store,
CommonProto::ReadConn { .from = conn.from });
return CommonProto::Serialise<T>::read(
CommonProto::ReadConn{.from = conn.from, .store = conn.store}
);
}
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WorkerProto::WriteConn conn, const T & t)
static WireFormatGenerator write(WorkerProto::WriteConn conn, const T & t)
{
return CommonProto::Serialise<T>::write(store,
CommonProto::WriteConn {},
t);
return CommonProto::Serialise<T>::write(CommonProto::WriteConn{.store = conn.store}, t);
}
};
lix/libstore/worker-protocol.cc
+34 -34
View File
@@ -11,7 +11,7 @@ namespace nix {
/* protocol-specific definitions */
std::optional<TrustedFlag> WorkerProto::Serialise<std::optional<TrustedFlag>>::read(const Store & store, WorkerProto::ReadConn conn)
std::optional<TrustedFlag> WorkerProto::Serialise<std::optional<TrustedFlag>>::read(WorkerProto::ReadConn conn)
{
auto temp = readNum<uint8_t>(conn.from);
switch (temp) {
@@ -26,7 +26,7 @@ std::optional<TrustedFlag> WorkerProto::Serialise<std::optional<TrustedFlag>>::r
}
}
WireFormatGenerator WorkerProto::Serialise<std::optional<TrustedFlag>>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional<TrustedFlag> & optTrusted)
WireFormatGenerator WorkerProto::Serialise<std::optional<TrustedFlag>>::write(WorkerProto::WriteConn conn, const std::optional<TrustedFlag> & optTrusted)
{
if (!optTrusted)
co_yield (uint8_t)0;
@@ -45,36 +45,36 @@ WireFormatGenerator WorkerProto::Serialise<std::optional<TrustedFlag>>::write(co
}
DerivedPath WorkerProto::Serialise<DerivedPath>::read(const Store & store, WorkerProto::ReadConn conn)
DerivedPath WorkerProto::Serialise<DerivedPath>::read(WorkerProto::ReadConn conn)
{
auto s = readString(conn.from);
return DerivedPath::parseLegacy(store, s);
return DerivedPath::parseLegacy(conn.store, s);
}
WireFormatGenerator WorkerProto::Serialise<DerivedPath>::write(const Store & store, WorkerProto::WriteConn conn, const DerivedPath & req)
WireFormatGenerator WorkerProto::Serialise<DerivedPath>::write(WorkerProto::WriteConn conn, const DerivedPath & req)
{
co_yield req.to_string_legacy(store);
co_yield req.to_string_legacy(conn.store);
}
KeyedBuildResult WorkerProto::Serialise<KeyedBuildResult>::read(const Store & store, WorkerProto::ReadConn conn)
KeyedBuildResult WorkerProto::Serialise<KeyedBuildResult>::read(WorkerProto::ReadConn conn)
{
auto path = WorkerProto::Serialise<DerivedPath>::read(store, conn);
auto br = WorkerProto::Serialise<BuildResult>::read(store, conn);
auto path = WorkerProto::Serialise<DerivedPath>::read(conn);
auto br = WorkerProto::Serialise<BuildResult>::read(conn);
return KeyedBuildResult {
std::move(br),
/* .path = */ std::move(path),
};
}
WireFormatGenerator WorkerProto::Serialise<KeyedBuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const KeyedBuildResult & res)
WireFormatGenerator WorkerProto::Serialise<KeyedBuildResult>::write(WorkerProto::WriteConn conn, const KeyedBuildResult & res)
{
co_yield WorkerProto::write(store, conn, res.path);
co_yield WorkerProto::write(store, conn, static_cast<const BuildResult &>(res));
co_yield WorkerProto::write(conn, res.path);
co_yield WorkerProto::write(conn, static_cast<const BuildResult &>(res));
}
BuildResult WorkerProto::Serialise<BuildResult>::read(const Store & store, WorkerProto::ReadConn conn)
BuildResult WorkerProto::Serialise<BuildResult>::read(WorkerProto::ReadConn conn)
{
BuildResult res;
res.status = (BuildResult::Status) readInt(conn.from);
@@ -84,7 +84,7 @@ BuildResult WorkerProto::Serialise<BuildResult>::read(const Store & store, Worke
>> res.isNonDeterministic
>> res.startTime
>> res.stopTime;
auto builtOutputs = WorkerProto::Serialise<DrvOutputs>::read(store, conn);
auto builtOutputs = WorkerProto::Serialise<DrvOutputs>::read(conn);
for (auto && [output, realisation] : builtOutputs)
res.builtOutputs.insert_or_assign(
std::move(output.outputName),
@@ -92,7 +92,7 @@ BuildResult WorkerProto::Serialise<BuildResult>::read(const Store & store, Worke
return res;
}
WireFormatGenerator WorkerProto::Serialise<BuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const BuildResult & res)
WireFormatGenerator WorkerProto::Serialise<BuildResult>::write(WorkerProto::WriteConn conn, const BuildResult & res)
{
co_yield res.status;
co_yield res.errorMsg;
@@ -103,33 +103,33 @@ WireFormatGenerator WorkerProto::Serialise<BuildResult>::write(const Store & sto
DrvOutputs builtOutputs;
for (auto & [output, realisation] : res.builtOutputs)
builtOutputs.insert_or_assign(realisation.id, realisation);
co_yield WorkerProto::write(store, conn, builtOutputs);
co_yield WorkerProto::write(conn, builtOutputs);
}
ValidPathInfo WorkerProto::Serialise<ValidPathInfo>::read(const Store & store, ReadConn conn)
ValidPathInfo WorkerProto::Serialise<ValidPathInfo>::read(ReadConn conn)
{
auto path = WorkerProto::Serialise<StorePath>::read(store, conn);
auto path = WorkerProto::Serialise<StorePath>::read(conn);
return ValidPathInfo {
std::move(path),
WorkerProto::Serialise<UnkeyedValidPathInfo>::read(store, conn),
WorkerProto::Serialise<UnkeyedValidPathInfo>::read(conn),
};
}
WireFormatGenerator WorkerProto::Serialise<ValidPathInfo>::write(const Store & store, WriteConn conn, const ValidPathInfo & pathInfo)
WireFormatGenerator WorkerProto::Serialise<ValidPathInfo>::write(WriteConn conn, const ValidPathInfo & pathInfo)
{
co_yield WorkerProto::write(store, conn, pathInfo.path);
co_yield WorkerProto::write(store, conn, static_cast<const UnkeyedValidPathInfo &>(pathInfo));
co_yield WorkerProto::write(conn, pathInfo.path);
co_yield WorkerProto::write(conn, static_cast<const UnkeyedValidPathInfo &>(pathInfo));
}
UnkeyedValidPathInfo WorkerProto::Serialise<UnkeyedValidPathInfo>::read(const Store & store, ReadConn conn)
UnkeyedValidPathInfo WorkerProto::Serialise<UnkeyedValidPathInfo>::read(ReadConn conn)
{
auto deriver = readString(conn.from);
auto narHash = Hash::parseAny(readString(conn.from), HashType::SHA256);
UnkeyedValidPathInfo info(narHash);
if (deriver != "") info.deriver = store.parseStorePath(deriver);
info.references = WorkerProto::Serialise<StorePathSet>::read(store, conn);
if (deriver != "") info.deriver = conn.store.parseStorePath(deriver);
info.references = WorkerProto::Serialise<StorePathSet>::read(conn);
conn.from >> info.registrationTime >> info.narSize;
conn.from >> info.ultimate;
@@ -139,11 +139,11 @@ UnkeyedValidPathInfo WorkerProto::Serialise<UnkeyedValidPathInfo>::read(const St
return info;
}
WireFormatGenerator WorkerProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & pathInfo)
WireFormatGenerator WorkerProto::Serialise<UnkeyedValidPathInfo>::write(WriteConn conn, const UnkeyedValidPathInfo & pathInfo)
{
co_yield (pathInfo.deriver ? store.printStorePath(*pathInfo.deriver) : "");
co_yield (pathInfo.deriver ? conn.store.printStorePath(*pathInfo.deriver) : "");
co_yield pathInfo.narHash.to_string(Base::Base16, false);
co_yield WorkerProto::write(store, conn, pathInfo.references);
co_yield WorkerProto::write(conn, pathInfo.references);
co_yield pathInfo.registrationTime;
co_yield pathInfo.narSize;
@@ -153,22 +153,22 @@ WireFormatGenerator WorkerProto::Serialise<UnkeyedValidPathInfo>::write(const St
}
SubstitutablePathInfo WorkerProto::Serialise<SubstitutablePathInfo>::read(const Store & store, ReadConn conn)
SubstitutablePathInfo WorkerProto::Serialise<SubstitutablePathInfo>::read(ReadConn conn)
{
SubstitutablePathInfo info;
auto deriver = readString(conn.from);
if (deriver != "")
info.deriver = store.parseStorePath(deriver);
info.references = WorkerProto::Serialise<StorePathSet>::read(store, conn);
info.deriver = conn.store.parseStorePath(deriver);
info.references = WorkerProto::Serialise<StorePathSet>::read(conn);
info.downloadSize = readLongLong(conn.from);
info.narSize = readLongLong(conn.from);
return info;
}
WireFormatGenerator WorkerProto::Serialise<SubstitutablePathInfo>::write(const Store & store, WriteConn conn, const SubstitutablePathInfo & info)
WireFormatGenerator WorkerProto::Serialise<SubstitutablePathInfo>::write(WriteConn conn, const SubstitutablePathInfo & info)
{
co_yield (info.deriver ? store.printStorePath(*info.deriver) : "");
co_yield WorkerProto::write(store, conn, info.references);
co_yield (info.deriver ? conn.store.printStorePath(*info.deriver) : "");
co_yield WorkerProto::write(conn, info.references);
co_yield info.downloadSize;
co_yield info.narSize;
}
lix/libstore/worker-protocol.hh
+15 -8
View File
@@ -72,9 +72,14 @@ struct WorkerProto
*/
struct ReadConn {
Source & from;
const Store & store;
Version version;
ReadConn(Source & from, Version version) : from(from), version(version) {
ReadConn(Source & from, const Store & store, Version version)
: from(from)
, store(store)
, version(version)
{
assert(version >= MIN_SUPPORTED_WORKER_PROTO_VERSION);
}
};
@@ -84,9 +89,11 @@ struct WorkerProto
* canonical serializers below.
*/
struct WriteConn {
const Store & store;
Version version;
explicit WriteConn(Version version) : version(version) {
WriteConn(const Store & store, Version version) : store(store), version(version)
{
assert(version >= MIN_SUPPORTED_WORKER_PROTO_VERSION);
}
};
@@ -117,8 +124,8 @@ struct WorkerProto
// This makes for a quicker debug cycle, as desired.
#if 0
{
static T read(const Store & store, ReadConn conn);
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t);
static T read(ReadConn conn);
static WireFormatGenerator write(WriteConn conn, const T & t);
};
#endif
@@ -128,9 +135,9 @@ struct WorkerProto
*/
template<typename T>
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
static WireFormatGenerator write(WriteConn conn, const T & t)
{
return WorkerProto::Serialise<T>::write(store, conn, t);
return WorkerProto::Serialise<T>::write(conn, t);
}
};
@@ -215,8 +222,8 @@ inline std::ostream & operator << (std::ostream & s, WorkerProto::Op op)
#define DECLARE_WORKER_SERIALISER(T) \
struct WorkerProto::Serialise< T > \
{ \
static T read(const Store & store, WorkerProto::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, WorkerProto::WriteConn conn, const T & t); \
static T read(WorkerProto::ReadConn conn); \
[[nodiscard]] static WireFormatGenerator write(WorkerProto::WriteConn conn, const T & t); \
};
template<>
tests/unit/libstore/common-protocol.cc
+2 -6
View File
@@ -32,8 +32,7 @@ public:
T got = ({
StringSource from { encoded };
CommonProto::Serialise<T>::read(
*store,
CommonProto::ReadConn { .from = from });
CommonProto::ReadConn { .from = from, .store = *store });
});
ASSERT_EQ(got, value);
@@ -49,10 +48,7 @@ public:
auto file = goldenMaster(testStem);
StringSink to;
to << CommonProto::write(
*store,
CommonProto::WriteConn {},
value);
to << CommonProto::write(CommonProto::WriteConn{*store}, value);
if (testAccept())
{
tests/unit/libstore/protocol.hh
+2 -6
View File
@@ -40,8 +40,7 @@ public:
T got = ({
StringSource from { expected };
Proto::template Serialise<T>::read(
*LibStoreTest::store,
typename Proto::ReadConn {from, version}
typename Proto::ReadConn{from, *LibStoreTest::store, version}
);
});
@@ -58,10 +57,7 @@ public:
auto file = ProtoTest<Proto, protocolDir>::goldenMaster(testStem);
StringSink to;
to << Proto::write(
*LibStoreTest::store,
typename Proto::WriteConn {version},
value);
to << Proto::write(typename Proto::WriteConn{*LibStoreTest::store, version}, value);
if (testAccept())
{