mirror of
https://github.com/privatevoid-net/nix-super.git
synced 2025-02-18 08:07:17 +02:00
Factor out commonality between WorkerProto::Basic{Client,Server}Connection
This also renames clientVersion and daemonVersion to the more correct protoVersion (since it's the version agreed to by both sides).
This commit is contained in:
parent
8ce4287409
commit
c1d5cf6f34
8 changed files with 100 additions and 130 deletions
|
@ -1020,8 +1020,8 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
|
||||||
|
|
||||||
void processConnection(
|
void processConnection(
|
||||||
ref<Store> store,
|
ref<Store> store,
|
||||||
FdSource & from,
|
FdSource && from,
|
||||||
FdSink & to,
|
FdSink && to,
|
||||||
TrustedFlag trusted,
|
TrustedFlag trusted,
|
||||||
RecursiveFlag recursive)
|
RecursiveFlag recursive)
|
||||||
{
|
{
|
||||||
|
@ -1037,7 +1037,12 @@ void processConnection(
|
||||||
if (clientVersion < 0x10a)
|
if (clientVersion < 0x10a)
|
||||||
throw Error("the Nix client version is too old");
|
throw Error("the Nix client version is too old");
|
||||||
|
|
||||||
auto tunnelLogger = new TunnelLogger(to, clientVersion);
|
WorkerProto::BasicServerConnection conn;
|
||||||
|
conn.to = std::move(to);
|
||||||
|
conn.from = std::move(from);
|
||||||
|
conn.protoVersion = clientVersion;
|
||||||
|
|
||||||
|
auto tunnelLogger = new TunnelLogger(conn.to, clientVersion);
|
||||||
auto prevLogger = nix::logger;
|
auto prevLogger = nix::logger;
|
||||||
// FIXME
|
// FIXME
|
||||||
if (!recursive)
|
if (!recursive)
|
||||||
|
@ -1050,12 +1055,6 @@ void processConnection(
|
||||||
printMsgUsing(prevLogger, lvlDebug, "%d operations", opCount);
|
printMsgUsing(prevLogger, lvlDebug, "%d operations", opCount);
|
||||||
});
|
});
|
||||||
|
|
||||||
WorkerProto::BasicServerConnection conn {
|
|
||||||
.to = to,
|
|
||||||
.from = from,
|
|
||||||
.clientVersion = clientVersion,
|
|
||||||
};
|
|
||||||
|
|
||||||
conn.postHandshake(*store, {
|
conn.postHandshake(*store, {
|
||||||
.daemonNixVersion = nixVersion,
|
.daemonNixVersion = nixVersion,
|
||||||
// We and the underlying store both need to trust the client for
|
// We and the underlying store both need to trust the client for
|
||||||
|
@ -1071,13 +1070,13 @@ void processConnection(
|
||||||
try {
|
try {
|
||||||
|
|
||||||
tunnelLogger->stopWork();
|
tunnelLogger->stopWork();
|
||||||
to.flush();
|
conn.to.flush();
|
||||||
|
|
||||||
/* Process client requests. */
|
/* Process client requests. */
|
||||||
while (true) {
|
while (true) {
|
||||||
WorkerProto::Op op;
|
WorkerProto::Op op;
|
||||||
try {
|
try {
|
||||||
op = (enum WorkerProto::Op) readInt(from);
|
op = (enum WorkerProto::Op) readInt(conn.from);
|
||||||
} catch (Interrupted & e) {
|
} catch (Interrupted & e) {
|
||||||
break;
|
break;
|
||||||
} catch (EndOfFile & e) {
|
} catch (EndOfFile & e) {
|
||||||
|
@ -1091,7 +1090,7 @@ void processConnection(
|
||||||
debug("performing daemon worker op: %d", op);
|
debug("performing daemon worker op: %d", op);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
performOp(tunnelLogger, store, trusted, recursive, clientVersion, from, to, op);
|
performOp(tunnelLogger, store, trusted, recursive, clientVersion, conn.from, conn.to, op);
|
||||||
} catch (Error & e) {
|
} catch (Error & e) {
|
||||||
/* If we're not in a state where we can send replies, then
|
/* If we're not in a state where we can send replies, then
|
||||||
something went wrong processing the input of the
|
something went wrong processing the input of the
|
||||||
|
@ -1107,19 +1106,19 @@ void processConnection(
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
|
|
||||||
to.flush();
|
conn.to.flush();
|
||||||
|
|
||||||
assert(!tunnelLogger->state_.lock()->canSendStderr);
|
assert(!tunnelLogger->state_.lock()->canSendStderr);
|
||||||
};
|
};
|
||||||
|
|
||||||
} catch (Error & e) {
|
} catch (Error & e) {
|
||||||
tunnelLogger->stopWork(&e);
|
tunnelLogger->stopWork(&e);
|
||||||
to.flush();
|
conn.to.flush();
|
||||||
return;
|
return;
|
||||||
} catch (std::exception & e) {
|
} catch (std::exception & e) {
|
||||||
auto ex = Error(e.what());
|
auto ex = Error(e.what());
|
||||||
tunnelLogger->stopWork(&ex);
|
tunnelLogger->stopWork(&ex);
|
||||||
to.flush();
|
conn.to.flush();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -10,8 +10,8 @@ enum RecursiveFlag : bool { NotRecursive = false, Recursive = true };
|
||||||
|
|
||||||
void processConnection(
|
void processConnection(
|
||||||
ref<Store> store,
|
ref<Store> store,
|
||||||
FdSource & from,
|
FdSource && from,
|
||||||
FdSink & to,
|
FdSink && to,
|
||||||
TrustedFlag trusted,
|
TrustedFlag trusted,
|
||||||
RecursiveFlag recursive);
|
RecursiveFlag recursive);
|
||||||
|
|
||||||
|
|
|
@ -73,7 +73,7 @@ void RemoteStore::initConnection(Connection & conn)
|
||||||
StringSink saved;
|
StringSink saved;
|
||||||
TeeSource tee(conn.from, saved);
|
TeeSource tee(conn.from, saved);
|
||||||
try {
|
try {
|
||||||
conn.daemonVersion = WorkerProto::BasicClientConnection::handshake(
|
conn.protoVersion = WorkerProto::BasicClientConnection::handshake(
|
||||||
conn.to, tee, PROTOCOL_VERSION);
|
conn.to, tee, PROTOCOL_VERSION);
|
||||||
} catch (SerialisationError & e) {
|
} catch (SerialisationError & e) {
|
||||||
/* In case the other side is waiting for our input, close
|
/* In case the other side is waiting for our input, close
|
||||||
|
@ -115,7 +115,7 @@ void RemoteStore::setOptions(Connection & conn)
|
||||||
<< settings.buildCores
|
<< settings.buildCores
|
||||||
<< settings.useSubstitutes;
|
<< settings.useSubstitutes;
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 12) {
|
if (GET_PROTOCOL_MINOR(conn.protoVersion) >= 12) {
|
||||||
std::map<std::string, Config::SettingInfo> overrides;
|
std::map<std::string, Config::SettingInfo> overrides;
|
||||||
settings.getSettings(overrides, true); // libstore settings
|
settings.getSettings(overrides, true); // libstore settings
|
||||||
fileTransferSettings.getSettings(overrides, true);
|
fileTransferSettings.getSettings(overrides, true);
|
||||||
|
@ -175,7 +175,7 @@ bool RemoteStore::isValidPathUncached(const StorePath & path)
|
||||||
StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
|
StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
|
||||||
{
|
{
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 12) {
|
||||||
StorePathSet res;
|
StorePathSet res;
|
||||||
for (auto & i : paths)
|
for (auto & i : paths)
|
||||||
if (isValidPath(i)) res.insert(i);
|
if (isValidPath(i)) res.insert(i);
|
||||||
|
@ -198,7 +198,7 @@ StorePathSet RemoteStore::queryAllValidPaths()
|
||||||
StorePathSet RemoteStore::querySubstitutablePaths(const StorePathSet & paths)
|
StorePathSet RemoteStore::querySubstitutablePaths(const StorePathSet & paths)
|
||||||
{
|
{
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 12) {
|
||||||
StorePathSet res;
|
StorePathSet res;
|
||||||
for (auto & i : paths) {
|
for (auto & i : paths) {
|
||||||
conn->to << WorkerProto::Op::HasSubstitutes << printStorePath(i);
|
conn->to << WorkerProto::Op::HasSubstitutes << printStorePath(i);
|
||||||
|
@ -221,7 +221,7 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S
|
||||||
|
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 12) {
|
||||||
|
|
||||||
for (auto & i : pathsMap) {
|
for (auto & i : pathsMap) {
|
||||||
SubstitutablePathInfo info;
|
SubstitutablePathInfo info;
|
||||||
|
@ -241,7 +241,7 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S
|
||||||
} else {
|
} else {
|
||||||
|
|
||||||
conn->to << WorkerProto::Op::QuerySubstitutablePathInfos;
|
conn->to << WorkerProto::Op::QuerySubstitutablePathInfos;
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 22) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 22) {
|
||||||
StorePathSet paths;
|
StorePathSet paths;
|
||||||
for (auto & path : pathsMap)
|
for (auto & path : pathsMap)
|
||||||
paths.insert(path.first);
|
paths.insert(path.first);
|
||||||
|
@ -368,7 +368,7 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
|
||||||
std::optional<ConnectionHandle> conn_(getConnection());
|
std::optional<ConnectionHandle> conn_(getConnection());
|
||||||
auto & conn = *conn_;
|
auto & conn = *conn_;
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 25) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) >= 25) {
|
||||||
|
|
||||||
conn->to
|
conn->to
|
||||||
<< WorkerProto::Op::AddToStore
|
<< WorkerProto::Op::AddToStore
|
||||||
|
@ -485,7 +485,7 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
|
||||||
{
|
{
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 18) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 18) {
|
||||||
auto source2 = sinkToSource([&](Sink & sink) {
|
auto source2 = sinkToSource([&](Sink & sink) {
|
||||||
sink << 1 // == path follows
|
sink << 1 // == path follows
|
||||||
;
|
;
|
||||||
|
@ -513,11 +513,11 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
|
||||||
<< info.ultimate << info.sigs << renderContentAddress(info.ca)
|
<< info.ultimate << info.sigs << renderContentAddress(info.ca)
|
||||||
<< repair << !checkSigs;
|
<< repair << !checkSigs;
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 23) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) >= 23) {
|
||||||
conn.withFramedSink([&](Sink & sink) {
|
conn.withFramedSink([&](Sink & sink) {
|
||||||
copyNAR(source, sink);
|
copyNAR(source, sink);
|
||||||
});
|
});
|
||||||
} else if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 21) {
|
} else if (GET_PROTOCOL_MINOR(conn->protoVersion) >= 21) {
|
||||||
conn.processStderr(0, &source);
|
conn.processStderr(0, &source);
|
||||||
} else {
|
} else {
|
||||||
copyNAR(source, conn->to);
|
copyNAR(source, conn->to);
|
||||||
|
@ -554,7 +554,7 @@ void RemoteStore::addMultipleToStore(
|
||||||
RepairFlag repair,
|
RepairFlag repair,
|
||||||
CheckSigsFlag checkSigs)
|
CheckSigsFlag checkSigs)
|
||||||
{
|
{
|
||||||
if (GET_PROTOCOL_MINOR(getConnection()->daemonVersion) >= 32) {
|
if (GET_PROTOCOL_MINOR(getConnection()->protoVersion) >= 32) {
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
conn->to
|
conn->to
|
||||||
<< WorkerProto::Op::AddMultipleToStore
|
<< WorkerProto::Op::AddMultipleToStore
|
||||||
|
@ -572,7 +572,7 @@ void RemoteStore::registerDrvOutput(const Realisation & info)
|
||||||
{
|
{
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
conn->to << WorkerProto::Op::RegisterDrvOutput;
|
conn->to << WorkerProto::Op::RegisterDrvOutput;
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 31) {
|
||||||
conn->to << info.id.to_string();
|
conn->to << info.id.to_string();
|
||||||
conn->to << std::string(info.outPath.to_string());
|
conn->to << std::string(info.outPath.to_string());
|
||||||
} else {
|
} else {
|
||||||
|
@ -587,7 +587,7 @@ void RemoteStore::queryRealisationUncached(const DrvOutput & id,
|
||||||
try {
|
try {
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 27) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 27) {
|
||||||
warn("the daemon is too old to support content-addressed derivations, please upgrade it to 2.4");
|
warn("the daemon is too old to support content-addressed derivations, please upgrade it to 2.4");
|
||||||
return callback(nullptr);
|
return callback(nullptr);
|
||||||
}
|
}
|
||||||
|
@ -597,7 +597,7 @@ void RemoteStore::queryRealisationUncached(const DrvOutput & id,
|
||||||
conn.processStderr();
|
conn.processStderr();
|
||||||
|
|
||||||
auto real = [&]() -> std::shared_ptr<const Realisation> {
|
auto real = [&]() -> std::shared_ptr<const Realisation> {
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 31) {
|
||||||
auto outPaths = WorkerProto::Serialise<std::set<StorePath>>::read(
|
auto outPaths = WorkerProto::Serialise<std::set<StorePath>>::read(
|
||||||
*this, *conn);
|
*this, *conn);
|
||||||
if (outPaths.empty())
|
if (outPaths.empty())
|
||||||
|
@ -644,9 +644,9 @@ void RemoteStore::buildPaths(const std::vector<DerivedPath> & drvPaths, BuildMod
|
||||||
|
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
conn->to << WorkerProto::Op::BuildPaths;
|
conn->to << WorkerProto::Op::BuildPaths;
|
||||||
assert(GET_PROTOCOL_MINOR(conn->daemonVersion) >= 13);
|
assert(GET_PROTOCOL_MINOR(conn->protoVersion) >= 13);
|
||||||
WorkerProto::write(*this, *conn, drvPaths);
|
WorkerProto::write(*this, *conn, drvPaths);
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 15)
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) >= 15)
|
||||||
conn->to << buildMode;
|
conn->to << buildMode;
|
||||||
else
|
else
|
||||||
/* Old daemons did not take a 'buildMode' parameter, so we
|
/* Old daemons did not take a 'buildMode' parameter, so we
|
||||||
|
@ -667,7 +667,7 @@ std::vector<KeyedBuildResult> RemoteStore::buildPathsWithResults(
|
||||||
std::optional<ConnectionHandle> conn_(getConnection());
|
std::optional<ConnectionHandle> conn_(getConnection());
|
||||||
auto & conn = *conn_;
|
auto & conn = *conn_;
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) {
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) >= 34) {
|
||||||
conn->to << WorkerProto::Op::BuildPathsWithResults;
|
conn->to << WorkerProto::Op::BuildPathsWithResults;
|
||||||
WorkerProto::write(*this, *conn, paths);
|
WorkerProto::write(*this, *conn, paths);
|
||||||
conn->to << buildMode;
|
conn->to << buildMode;
|
||||||
|
@ -841,7 +841,7 @@ void RemoteStore::queryMissing(const std::vector<DerivedPath> & targets,
|
||||||
{
|
{
|
||||||
{
|
{
|
||||||
auto conn(getConnection());
|
auto conn(getConnection());
|
||||||
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 19)
|
if (GET_PROTOCOL_MINOR(conn->protoVersion) < 19)
|
||||||
// Don't hold the connection handle in the fallback case
|
// Don't hold the connection handle in the fallback case
|
||||||
// to prevent a deadlock.
|
// to prevent a deadlock.
|
||||||
goto fallback;
|
goto fallback;
|
||||||
|
@ -889,7 +889,7 @@ void RemoteStore::connect()
|
||||||
unsigned int RemoteStore::getProtocol()
|
unsigned int RemoteStore::getProtocol()
|
||||||
{
|
{
|
||||||
auto conn(connections->get());
|
auto conn(connections->get());
|
||||||
return conn->daemonVersion;
|
return conn->protoVersion;
|
||||||
}
|
}
|
||||||
|
|
||||||
std::optional<TrustedFlag> RemoteStore::isTrustedClient()
|
std::optional<TrustedFlag> RemoteStore::isTrustedClient()
|
||||||
|
|
|
@ -1526,10 +1526,11 @@ void LocalDerivationGoal::startDaemon()
|
||||||
debug("received daemon connection");
|
debug("received daemon connection");
|
||||||
|
|
||||||
auto workerThread = std::thread([store, remote{std::move(remote)}]() {
|
auto workerThread = std::thread([store, remote{std::move(remote)}]() {
|
||||||
FdSource from(remote.get());
|
|
||||||
FdSink to(remote.get());
|
|
||||||
try {
|
try {
|
||||||
daemon::processConnection(store, from, to,
|
daemon::processConnection(
|
||||||
|
store,
|
||||||
|
FdSource(remote.get()),
|
||||||
|
FdSink(remote.get()),
|
||||||
NotTrusted, daemon::Recursive);
|
NotTrusted, daemon::Recursive);
|
||||||
debug("terminated daemon connection");
|
debug("terminated daemon connection");
|
||||||
} catch (SystemError &) {
|
} catch (SystemError &) {
|
||||||
|
|
|
@ -58,7 +58,7 @@ std::exception_ptr WorkerProto::BasicClientConnection::processStderrReturn(Sink
|
||||||
}
|
}
|
||||||
|
|
||||||
else if (msg == STDERR_ERROR) {
|
else if (msg == STDERR_ERROR) {
|
||||||
if (GET_PROTOCOL_MINOR(daemonVersion) >= 26) {
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 26) {
|
||||||
ex = std::make_exception_ptr(readError(from));
|
ex = std::make_exception_ptr(readError(from));
|
||||||
} else {
|
} else {
|
||||||
auto error = readString(from);
|
auto error = readString(from);
|
||||||
|
@ -114,7 +114,7 @@ std::exception_ptr WorkerProto::BasicClientConnection::processStderrReturn(Sink
|
||||||
// explain to users what's going on when their daemon is
|
// explain to users what's going on when their daemon is
|
||||||
// older than #4628 (2023).
|
// older than #4628 (2023).
|
||||||
if (experimentalFeatureSettings.isEnabled(Xp::DynamicDerivations)
|
if (experimentalFeatureSettings.isEnabled(Xp::DynamicDerivations)
|
||||||
&& GET_PROTOCOL_MINOR(daemonVersion) <= 35) {
|
&& GET_PROTOCOL_MINOR(protoVersion) <= 35) {
|
||||||
auto m = e.msg();
|
auto m = e.msg();
|
||||||
if (m.find("parsing derivation") != std::string::npos && m.find("expected string") != std::string::npos
|
if (m.find("parsing derivation") != std::string::npos && m.find("expected string") != std::string::npos
|
||||||
&& m.find("Derive([") != std::string::npos)
|
&& m.find("Derive([") != std::string::npos)
|
||||||
|
@ -172,15 +172,15 @@ WorkerProto::ClientHandshakeInfo WorkerProto::BasicClientConnection::postHandsha
|
||||||
{
|
{
|
||||||
WorkerProto::ClientHandshakeInfo res;
|
WorkerProto::ClientHandshakeInfo res;
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(daemonVersion) >= 14) {
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 14) {
|
||||||
// Obsolete CPU affinity.
|
// Obsolete CPU affinity.
|
||||||
to << 0;
|
to << 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(daemonVersion) >= 11)
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 11)
|
||||||
to << false; // obsolete reserveSpace
|
to << false; // obsolete reserveSpace
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(daemonVersion) >= 33)
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 33)
|
||||||
to.flush();
|
to.flush();
|
||||||
|
|
||||||
return WorkerProto::Serialise<ClientHandshakeInfo>::read(store, *this);
|
return WorkerProto::Serialise<ClientHandshakeInfo>::read(store, *this);
|
||||||
|
@ -188,12 +188,12 @@ WorkerProto::ClientHandshakeInfo WorkerProto::BasicClientConnection::postHandsha
|
||||||
|
|
||||||
void WorkerProto::BasicServerConnection::postHandshake(const StoreDirConfig & store, const ClientHandshakeInfo & info)
|
void WorkerProto::BasicServerConnection::postHandshake(const StoreDirConfig & store, const ClientHandshakeInfo & info)
|
||||||
{
|
{
|
||||||
if (GET_PROTOCOL_MINOR(clientVersion) >= 14 && readInt(from)) {
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 14 && readInt(from)) {
|
||||||
// Obsolete CPU affinity.
|
// Obsolete CPU affinity.
|
||||||
readInt(from);
|
readInt(from);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (GET_PROTOCOL_MINOR(clientVersion) >= 11)
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 11)
|
||||||
readInt(from); // obsolete reserveSpace
|
readInt(from); // obsolete reserveSpace
|
||||||
|
|
||||||
WorkerProto::write(store, *this, info);
|
WorkerProto::write(store, *this, info);
|
||||||
|
@ -211,7 +211,7 @@ UnkeyedValidPathInfo WorkerProto::BasicClientConnection::queryPathInfo(
|
||||||
throw InvalidPath(std::move(e.info()));
|
throw InvalidPath(std::move(e.info()));
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
if (GET_PROTOCOL_MINOR(daemonVersion) >= 17) {
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 17) {
|
||||||
bool valid;
|
bool valid;
|
||||||
from >> valid;
|
from >> valid;
|
||||||
if (!valid)
|
if (!valid)
|
||||||
|
@ -223,10 +223,10 @@ UnkeyedValidPathInfo WorkerProto::BasicClientConnection::queryPathInfo(
|
||||||
StorePathSet WorkerProto::BasicClientConnection::queryValidPaths(
|
StorePathSet WorkerProto::BasicClientConnection::queryValidPaths(
|
||||||
const StoreDirConfig & store, bool * daemonException, const StorePathSet & paths, SubstituteFlag maybeSubstitute)
|
const StoreDirConfig & store, bool * daemonException, const StorePathSet & paths, SubstituteFlag maybeSubstitute)
|
||||||
{
|
{
|
||||||
assert(GET_PROTOCOL_MINOR(daemonVersion) >= 12);
|
assert(GET_PROTOCOL_MINOR(protoVersion) >= 12);
|
||||||
to << WorkerProto::Op::QueryValidPaths;
|
to << WorkerProto::Op::QueryValidPaths;
|
||||||
WorkerProto::write(store, *this, paths);
|
WorkerProto::write(store, *this, paths);
|
||||||
if (GET_PROTOCOL_MINOR(daemonVersion) >= 27) {
|
if (GET_PROTOCOL_MINOR(protoVersion) >= 27) {
|
||||||
to << maybeSubstitute;
|
to << maybeSubstitute;
|
||||||
}
|
}
|
||||||
processStderr(daemonException);
|
processStderr(daemonException);
|
||||||
|
|
|
@ -6,7 +6,7 @@
|
||||||
|
|
||||||
namespace nix {
|
namespace nix {
|
||||||
|
|
||||||
struct WorkerProto::BasicClientConnection
|
struct WorkerProto::BasicConnection
|
||||||
{
|
{
|
||||||
/**
|
/**
|
||||||
* Send with this.
|
* Send with this.
|
||||||
|
@ -19,14 +19,45 @@ struct WorkerProto::BasicClientConnection
|
||||||
FdSource from;
|
FdSource from;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Worker protocol version used for the connection.
|
* The protocol version agreed by both sides.
|
||||||
*
|
|
||||||
* Despite its name, 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.)
|
|
||||||
*/
|
*/
|
||||||
WorkerProto::Version daemonVersion;
|
WorkerProto::Version protoVersion;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Coercion to `WorkerProto::ReadConn`. This makes it easy to use the
|
||||||
|
* factored out serve protocol serializers with a
|
||||||
|
* `LegacySSHStore::Connection`.
|
||||||
|
*
|
||||||
|
* The serve protocol connection types are unidirectional, unlike
|
||||||
|
* this type.
|
||||||
|
*/
|
||||||
|
operator WorkerProto::ReadConn()
|
||||||
|
{
|
||||||
|
return WorkerProto::ReadConn{
|
||||||
|
.from = from,
|
||||||
|
.version = protoVersion,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Coercion to `WorkerProto::WriteConn`. This makes it easy to use the
|
||||||
|
* factored out serve protocol serializers with a
|
||||||
|
* `LegacySSHStore::Connection`.
|
||||||
|
*
|
||||||
|
* The serve protocol connection types are unidirectional, unlike
|
||||||
|
* this type.
|
||||||
|
*/
|
||||||
|
operator WorkerProto::WriteConn()
|
||||||
|
{
|
||||||
|
return WorkerProto::WriteConn{
|
||||||
|
.to = to,
|
||||||
|
.version = protoVersion,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
struct WorkerProto::BasicClientConnection : WorkerProto::BasicConnection
|
||||||
|
{
|
||||||
/**
|
/**
|
||||||
* Flush to direction
|
* Flush to direction
|
||||||
*/
|
*/
|
||||||
|
@ -60,38 +91,6 @@ struct WorkerProto::BasicClientConnection
|
||||||
*/
|
*/
|
||||||
ClientHandshakeInfo postHandshake(const StoreDirConfig & store);
|
ClientHandshakeInfo postHandshake(const StoreDirConfig & store);
|
||||||
|
|
||||||
/**
|
|
||||||
* Coercion to `WorkerProto::ReadConn`. This makes it easy to use the
|
|
||||||
* factored out serve protocol serializers with a
|
|
||||||
* `LegacySSHStore::Connection`.
|
|
||||||
*
|
|
||||||
* The serve protocol connection types are unidirectional, unlike
|
|
||||||
* this type.
|
|
||||||
*/
|
|
||||||
operator WorkerProto::ReadConn()
|
|
||||||
{
|
|
||||||
return WorkerProto::ReadConn{
|
|
||||||
.from = from,
|
|
||||||
.version = daemonVersion,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Coercion to `WorkerProto::WriteConn`. This makes it easy to use the
|
|
||||||
* factored out serve protocol serializers with a
|
|
||||||
* `LegacySSHStore::Connection`.
|
|
||||||
*
|
|
||||||
* The serve protocol connection types are unidirectional, unlike
|
|
||||||
* this type.
|
|
||||||
*/
|
|
||||||
operator WorkerProto::WriteConn()
|
|
||||||
{
|
|
||||||
return WorkerProto::WriteConn{
|
|
||||||
.to = to,
|
|
||||||
.version = daemonVersion,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
void addTempRoot(const StoreDirConfig & remoteStore, bool * daemonException, const StorePath & path);
|
void addTempRoot(const StoreDirConfig & remoteStore, bool * daemonException, const StorePath & path);
|
||||||
|
|
||||||
StorePathSet queryValidPaths(
|
StorePathSet queryValidPaths(
|
||||||
|
@ -124,43 +123,8 @@ struct WorkerProto::BasicClientConnection
|
||||||
void importPaths(const StoreDirConfig & store, bool * daemonException, Source & source);
|
void importPaths(const StoreDirConfig & store, bool * daemonException, Source & source);
|
||||||
};
|
};
|
||||||
|
|
||||||
struct WorkerProto::BasicServerConnection
|
struct WorkerProto::BasicServerConnection : WorkerProto::BasicConnection
|
||||||
{
|
{
|
||||||
/**
|
|
||||||
* Send with this.
|
|
||||||
*/
|
|
||||||
FdSink & to;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Receive with this.
|
|
||||||
*/
|
|
||||||
FdSource & from;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Worker protocol version used for the connection.
|
|
||||||
*
|
|
||||||
* Despite its name, 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.)
|
|
||||||
*/
|
|
||||||
WorkerProto::Version clientVersion;
|
|
||||||
|
|
||||||
operator WorkerProto::ReadConn()
|
|
||||||
{
|
|
||||||
return WorkerProto::ReadConn{
|
|
||||||
.from = from,
|
|
||||||
.version = clientVersion,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
operator WorkerProto::WriteConn()
|
|
||||||
{
|
|
||||||
return WorkerProto::WriteConn{
|
|
||||||
.to = to,
|
|
||||||
.version = clientVersion,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Establishes connection, negotiating version.
|
* Establishes connection, negotiating version.
|
||||||
*
|
*
|
||||||
|
|
|
@ -82,6 +82,7 @@ struct WorkerProto
|
||||||
*
|
*
|
||||||
* @todo remove once Hydra uses Store abstraction consistently.
|
* @todo remove once Hydra uses Store abstraction consistently.
|
||||||
*/
|
*/
|
||||||
|
struct BasicConnection;
|
||||||
struct BasicClientConnection;
|
struct BasicClientConnection;
|
||||||
struct BasicServerConnection;
|
struct BasicServerConnection;
|
||||||
|
|
||||||
|
|
|
@ -370,9 +370,12 @@ static void daemonLoop(std::optional<TrustedFlag> forceTrustClientOpt)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Handle the connection.
|
// Handle the connection.
|
||||||
FdSource from(remote.get());
|
processConnection(
|
||||||
FdSink to(remote.get());
|
openUncachedStore(),
|
||||||
processConnection(openUncachedStore(), from, to, trusted, NotRecursive);
|
FdSource(remote.get()),
|
||||||
|
FdSink(remote.get()),
|
||||||
|
trusted,
|
||||||
|
NotRecursive);
|
||||||
|
|
||||||
exit(0);
|
exit(0);
|
||||||
}, options);
|
}, options);
|
||||||
|
@ -437,9 +440,11 @@ static void forwardStdioConnection(RemoteStore & store) {
|
||||||
*/
|
*/
|
||||||
static void processStdioConnection(ref<Store> store, TrustedFlag trustClient)
|
static void processStdioConnection(ref<Store> store, TrustedFlag trustClient)
|
||||||
{
|
{
|
||||||
FdSource from(STDIN_FILENO);
|
processConnection(
|
||||||
FdSink to(STDOUT_FILENO);
|
store,
|
||||||
processConnection(store, from, to, trustClient, NotRecursive);
|
FdSource(STDIN_FILENO),
|
||||||
|
FdSink(STDOUT_FILENO),
|
||||||
|
trustClient, NotRecursive);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
Loading…
Add table
Reference in a new issue