vriska/hydra
0
0
Fork 0
forked from lix-project/hydra
Code Pull requests Activity

Merge pull request #1341 from NixOS/machine-dedup

Browse source 
Use Nix's `Machine` type in a minimal way
This commit is contained in:
John Ericson 2024-01-23 15:38:19 -05:00 committed by GitHub
parent f5c0efb11e 70e5469303
commit d02e20a4c1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 69 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
src/hydra-queue-runner/build-remote.cc
View file

@ -48,7 +48,7 @@ static Strings extraStoreArgs(std::string & machine)
return result; return result;
} }
static void openConnection(Machine::ptr machine, Path tmpDir, int stderrFD, SSHMaster::Connection & child) static void openConnection(::Machine::ptr machine, Path tmpDir, int stderrFD, SSHMaster::Connection & child)
{ {
std::string pgmName; std::string pgmName;
Pipe to, from; Pipe to, from;
@ -104,7 +104,7 @@ static void openConnection(Machine::ptr machine, Path tmpDir, int stderrFD, SSHM
static void copyClosureTo( static void copyClosureTo(
Machine::Connection & conn, ::Machine::Connection & conn,
Store & destStore, Store & destStore,
const StorePathSet & paths, const StorePathSet & paths,
SubstituteFlag useSubstitutes = NoSubstitute) SubstituteFlag useSubstitutes = NoSubstitute)
@ -195,7 +195,7 @@ static std::pair<Path, AutoCloseFD> openLogFile(const std::string & logDir, cons
* Therefore, no `ServeProto::Serialize` functions can be used until * Therefore, no `ServeProto::Serialize` functions can be used until
* that field is set. * that field is set.
*/ */
static void handshake(Machine::Connection & conn, unsigned int repeats) static void handshake(::Machine::Connection & conn, unsigned int repeats)
{ {
conn.to << SERVE_MAGIC_1 << 0x206; conn.to << SERVE_MAGIC_1 << 0x206;
conn.to.flush(); conn.to.flush();
@ -216,7 +216,7 @@ static BasicDerivation sendInputs(
Step & step, Step & step,
Store & localStore, Store & localStore,
Store & destStore, Store & destStore,
Machine::Connection & conn, ::Machine::Connection & conn,
unsigned int & overhead, unsigned int & overhead,
counter & nrStepsWaiting, counter & nrStepsWaiting,
counter & nrStepsCopyingTo counter & nrStepsCopyingTo
@ -272,7 +272,7 @@ static BasicDerivation sendInputs(
} }
static BuildResult performBuild( static BuildResult performBuild(
Machine::Connection & conn, ::Machine::Connection & conn,
Store & localStore, Store & localStore,
StorePath drvPath, StorePath drvPath,
const BasicDerivation & drv, const BasicDerivation & drv,
@ -317,7 +317,7 @@ static BuildResult performBuild(
} }
static std::map<StorePath, ValidPathInfo> queryPathInfos( static std::map<StorePath, ValidPathInfo> queryPathInfos(
Machine::Connection & conn, ::Machine::Connection & conn,
Store & localStore, Store & localStore,
StorePathSet & outputs, StorePathSet & outputs,
size_t & totalNarSize size_t & totalNarSize
@ -355,7 +355,7 @@ static std::map<StorePath, ValidPathInfo> queryPathInfos(
} }
static void copyPathFromRemote( static void copyPathFromRemote(
Machine::Connection & conn, ::Machine::Connection & conn,
NarMemberDatas & narMembers, NarMemberDatas & narMembers,
Store & localStore, Store & localStore,
Store & destStore, Store & destStore,
@ -385,7 +385,7 @@ static void copyPathFromRemote(
} }
static void copyPathsFromRemote( static void copyPathsFromRemote(
Machine::Connection & conn, ::Machine::Connection & conn,
NarMemberDatas & narMembers, NarMemberDatas & narMembers,
Store & localStore, Store & localStore,
Store & destStore, Store & destStore,
@ -462,7 +462,7 @@ void RemoteResult::updateWithBuildResult(const nix::BuildResult & buildResult)
void State::buildRemote(ref<Store> destStore, void State::buildRemote(ref<Store> destStore,
Machine::ptr machine, Step::ptr step, ::Machine::ptr machine, Step::ptr step,
const BuildOptions & buildOptions, const BuildOptions & buildOptions,
RemoteResult & result, std::shared_ptr<ActiveStep> activeStep, RemoteResult & result, std::shared_ptr<ActiveStep> activeStep,
std::function<void(StepState)> updateStep, std::function<void(StepState)> updateStep,
@ -503,7 +503,7 @@ void State::buildRemote(ref<Store> destStore,
process. Meh. */ process. Meh. */
}); });
Machine::Connection conn { ::Machine::Connection conn {
.from = child.out.get(), .from = child.out.get(),
.to = child.in.get(), .to = child.in.get(),
.machine = machine, .machine = machine,

2
src/hydra-queue-runner/builder.cc
View file

@ -400,7 +400,7 @@ void State::failStep(
Step::ptr step, Step::ptr step,
BuildID buildId, BuildID buildId,
const RemoteResult & result, const RemoteResult & result,
Machine::ptr machine, ::Machine::ptr machine,
bool & stepFinished) bool & stepFinished)
{ {
/* Register failure in the database for all Build objects that /* Register failure in the database for all Build objects that

10
src/hydra-queue-runner/dispatcher.cc
View file

@ -199,7 +199,7 @@ system_time State::doDispatch()
filter out temporarily disabled machines. */ filter out temporarily disabled machines. */
struct MachineInfo struct MachineInfo
{ {
Machine::ptr machine; ::Machine::ptr machine;
unsigned long currentJobs; unsigned long currentJobs;
}; };
std::vector<MachineInfo> machinesSorted; std::vector<MachineInfo> machinesSorted;
@ -231,11 +231,11 @@ system_time State::doDispatch()
sort(machinesSorted.begin(), machinesSorted.end(), sort(machinesSorted.begin(), machinesSorted.end(),
[](const MachineInfo & a, const MachineInfo & b) -> bool [](const MachineInfo & a, const MachineInfo & b) -> bool
{ {
float ta = std::round(a.currentJobs / a.machine->speedFactor); float ta = std::round(a.currentJobs / a.machine->speedFactorFloat);
float tb = std::round(b.currentJobs / b.machine->speedFactor); float tb = std::round(b.currentJobs / b.machine->speedFactorFloat);
return return
ta != tb ? ta < tb : ta != tb ? ta < tb :
a.machine->speedFactor != b.machine->speedFactor ? a.machine->speedFactor > b.machine->speedFactor : a.machine->speedFactorFloat != b.machine->speedFactorFloat ? a.machine->speedFactorFloat > b.machine->speedFactorFloat :
a.currentJobs > b.currentJobs; a.currentJobs > b.currentJobs;
}); });
@ -435,7 +435,7 @@ void Jobset::pruneSteps()
} }
State::MachineReservation::MachineReservation(State & state, Step::ptr step, Machine::ptr machine) State::MachineReservation::MachineReservation(State & state, Step::ptr step, ::Machine::ptr machine)
: state(state), step(step), machine(machine) : state(state), step(step), machine(machine)
{ {
machine->state->currentJobs++; machine->state->currentJobs++;

59
src/hydra-queue-runner/hydra-queue-runner.cc
View file

@ -1,6 +1,7 @@
#include <iostream> #include <iostream>
#include <thread> #include <thread>
#include <optional> #include <optional>
#include <type_traits>
#include <sys/types.h> #include <sys/types.h>
#include <sys/stat.h> #include <sys/stat.h>
@ -140,23 +141,43 @@ void State::parseMachines(const std::string & contents)
if (tokens.size() < 3) continue; if (tokens.size() < 3) continue;
tokens.resize(8); tokens.resize(8);
auto machine = std::make_shared<Machine>();
machine->sshName = tokens[0];
machine->systemTypes = tokenizeString<StringSet>(tokens[1], ",");
machine->sshKey = tokens[2] == "-" ? std::string("") : tokens[2];
if (tokens[3] != "")
machine->maxJobs = string2Int<decltype(machine->maxJobs)>(tokens[3]).value();
else
machine->maxJobs = 1;
machine->speedFactor = atof(tokens[4].c_str());
if (tokens[5] == "-") tokens[5] = ""; if (tokens[5] == "-") tokens[5] = "";
machine->supportedFeatures = tokenizeString<StringSet>(tokens[5], ","); auto supportedFeatures = tokenizeString<StringSet>(tokens[5], ",");
if (tokens[6] == "-") tokens[6] = ""; if (tokens[6] == "-") tokens[6] = "";
machine->mandatoryFeatures = tokenizeString<StringSet>(tokens[6], ","); auto mandatoryFeatures = tokenizeString<StringSet>(tokens[6], ",");
for (auto & f : machine->mandatoryFeatures)
machine->supportedFeatures.insert(f); for (auto & f : mandatoryFeatures)
if (tokens[7] != "" && tokens[7] != "-") supportedFeatures.insert(f);
machine->sshPublicHostKey = base64Decode(tokens[7]);
using MaxJobs = std::remove_const<decltype(nix::Machine::maxJobs)>::type;
auto machine = std::make_shared<::Machine>(nix::Machine {
// `storeUri`, not yet used
"",
// `systemTypes`, not yet used
{},
// `sshKey`
tokens[2] == "-" ? "" : tokens[2],
// `maxJobs`
tokens[3] != ""
? string2Int<MaxJobs>(tokens[3]).value()
: 1,
// `speedFactor`, not yet used
1,
// `supportedFeatures`
std::move(supportedFeatures),
// `mandatoryFeatures`
std::move(mandatoryFeatures),
// `sshPublicHostKey`
tokens[7] != "" && tokens[7] != "-"
? base64Decode(tokens[7])
: "",
});
machine->sshName = tokens[0];
machine->systemTypesSet = tokenizeString<StringSet>(tokens[1], ",");
machine->speedFactorFloat = atof(tokens[4].c_str());
/* Re-use the State object of the previous machine with the /* Re-use the State object of the previous machine with the
same name. */ same name. */
@ -166,7 +187,7 @@ void State::parseMachines(const std::string & contents)
else else
printMsg(lvlChatty, "updating machine ‘%1%’", machine->sshName); printMsg(lvlChatty, "updating machine ‘%1%’", machine->sshName);
machine->state = i == oldMachines.end() machine->state = i == oldMachines.end()
? std::make_shared<Machine::State>() ? std::make_shared<::Machine::State>()
: i->second->state; : i->second->state;
newMachines[machine->sshName] = machine; newMachines[machine->sshName] = machine;
} }
@ -175,9 +196,9 @@ void State::parseMachines(const std::string & contents)
if (newMachines.find(m.first) == newMachines.end()) { if (newMachines.find(m.first) == newMachines.end()) {
if (m.second->enabled) if (m.second->enabled)
printInfo("removing machine ‘%1%’", m.first); printInfo("removing machine ‘%1%’", m.first);
/* Add a disabled Machine object to make sure stats are /* Add a disabled ::Machine object to make sure stats are
maintained. */ maintained. */
auto machine = std::make_shared<Machine>(*(m.second)); auto machine = std::make_shared<::Machine>(*(m.second));
machine->enabled = false; machine->enabled = false;
newMachines[m.first] = machine; newMachines[m.first] = machine;
} }
@ -596,7 +617,7 @@ void State::dumpStatus(Connection & conn)
json machine = { json machine = {
{"enabled", m->enabled}, {"enabled", m->enabled},
{"systemTypes", m->systemTypes}, {"systemTypes", m->systemTypesSet},
{"supportedFeatures", m->supportedFeatures}, {"supportedFeatures", m->supportedFeatures},
{"mandatoryFeatures", m->mandatoryFeatures}, {"mandatoryFeatures", m->mandatoryFeatures},
{"nrStepsDone", s->nrStepsDone.load()}, {"nrStepsDone", s->nrStepsDone.load()},

21
src/hydra-queue-runner/state.hh
View file

@ -22,6 +22,7 @@
#include "sync.hh" #include "sync.hh"
#include "nar-extractor.hh" #include "nar-extractor.hh"
#include "serve-protocol.hh" #include "serve-protocol.hh"
#include "machines.hh"
typedef unsigned int BuildID; typedef unsigned int BuildID;
@ -234,17 +235,21 @@ void getDependents(Step::ptr step, std::set<Build::ptr> & builds, std::set<Step:
void visitDependencies(std::function<void(Step::ptr)> visitor, Step::ptr step); void visitDependencies(std::function<void(Step::ptr)> visitor, Step::ptr step);
struct Machine struct Machine : nix::Machine
{ {
typedef std::shared_ptr<Machine> ptr; typedef std::shared_ptr<Machine> ptr;
bool enabled{true}; /* TODO Get rid of: `nix::Machine::storeUri` is normalized in a way
we are not yet used to, but once we are, we don't need this. */
std::string sshName;
std::string sshName, sshKey; /* TODO Get rid once `nix::Machine::systemTypes` is a set not
std::set<std::string> systemTypes, supportedFeatures, mandatoryFeatures; vector. */
unsigned int maxJobs = 1; std::set<std::string> systemTypesSet;
float speedFactor = 1.0;
std::string sshPublicHostKey; /* TODO Get rid once `nix::Machine::systemTypes` is a `float` not
an `int`. */
float speedFactorFloat = 1.0;
struct State { struct State {
typedef std::shared_ptr<State> ptr; typedef std::shared_ptr<State> ptr;
@ -272,7 +277,7 @@ struct Machine
{ {
/* Check that this machine is of the type required by the /* Check that this machine is of the type required by the
step. */ step. */
if (!systemTypes.count(step->drv->platform == "builtin" ? nix::settings.thisSystem : step->drv->platform)) if (!systemTypesSet.count(step->drv->platform == "builtin" ? nix::settings.thisSystem : step->drv->platform))
return false; return false;
/* Check that the step requires all mandatory features of this /* Check that the step requires all mandatory features of this