Skip to content

Commit f0b7b37

Browse files
mergify[bot]mergify[bot]
authored
Merge pull request #12697 from NixOS/worker-abstraction
Make Goal code use abstractions over interations with Worker
2 parents 83ec817 + 36e5aa6 commit f0b7b37

6 files changed

Lines changed: 110 additions & 75 deletions

File tree

‎src/libstore/build/derivation-goal.cc‎

Lines changed: 27 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -142,8 +142,8 @@ Goal::Co DerivationGoal::init() {
142142
substitute. */
143143

144144
if (buildMode != bmNormal || !worker.evalStore.isValidPath(drvPath)) {
145-
addWaitee(upcast_goal(worker.makePathSubstitutionGoal(drvPath)));
146-
co_await Suspend{};
145+
Goals waitees{upcast_goal(worker.makePathSubstitutionGoal(drvPath))};
146+
co_await await(std::move(waitees));
147147
}
148148

149149
trace("loading derivation");
@@ -235,14 +235,16 @@ Goal::Co DerivationGoal::haveDerivation()
235235
}
236236
}
237237

238+
Goals waitees;
239+
238240
/* We are first going to try to create the invalid output paths
239241
through substitutes. If that doesn't work, we'll build
240242
them. */
241243
if (settings.useSubstitutes && drvOptions->substitutesAllowed())
242244
for (auto & [outputName, status] : initialOutputs) {
243245
if (!status.wanted) continue;
244246
if (!status.known)
245-
addWaitee(
247+
waitees.insert(
246248
upcast_goal(
247249
worker.makeDrvOutputSubstitutionGoal(
248250
DrvOutput{status.outputHash, outputName},
@@ -252,14 +254,14 @@ Goal::Co DerivationGoal::haveDerivation()
252254
);
253255
else {
254256
auto * cap = getDerivationCA(*drv);
255-
addWaitee(upcast_goal(worker.makePathSubstitutionGoal(
257+
waitees.insert(upcast_goal(worker.makePathSubstitutionGoal(
256258
status.known->path,
257259
buildMode == bmRepair ? Repair : NoRepair,
258260
cap ? std::optional { *cap } : std::nullopt)));
259261
}
260262
}
261263

262-
if (!waitees.empty()) co_await Suspend{}; /* to prevent hang (no wake-up event) */
264+
co_await await(std::move(waitees));
263265

264266
trace("all outputs substituted (maybe)");
265267

@@ -342,6 +344,8 @@ Goal::Co DerivationGoal::gaveUpOnSubstitution()
342344
is no need to restart. */
343345
needRestart = NeedRestartForMoreOutputs::BuildInProgressWillNotNeed;
344346

347+
Goals waitees;
348+
345349
std::map<ref<const SingleDerivedPath>, GoalPtr, value_comparison> inputGoals;
346350

347351
if (useDerivation) {
@@ -356,7 +360,7 @@ Goal::Co DerivationGoal::gaveUpOnSubstitution()
356360
},
357361
buildMode == bmRepair ? bmRepair : bmNormal);
358362
inputGoals.insert_or_assign(inputDrv, g);
359-
addWaitee(std::move(g));
363+
waitees.insert(std::move(g));
360364
}
361365
for (const auto & [outputName, childNode] : inputNode.childMap)
362366
addWaiteeDerivedPath(
@@ -397,10 +401,11 @@ Goal::Co DerivationGoal::gaveUpOnSubstitution()
397401
if (!settings.useSubstitutes)
398402
throw Error("dependency '%s' of '%s' does not exist, and substitution is disabled",
399403
worker.store.printStorePath(i), worker.store.printStorePath(drvPath));
400-
addWaitee(upcast_goal(worker.makePathSubstitutionGoal(i)));
404+
waitees.insert(upcast_goal(worker.makePathSubstitutionGoal(i)));
401405
}
402406

403-
if (!waitees.empty()) co_await Suspend{}; /* to prevent hang (no wake-up event) */
407+
co_await await(std::move(waitees));
408+
404409

405410
trace("all inputs realised");
406411

@@ -495,9 +500,11 @@ Goal::Co DerivationGoal::gaveUpOnSubstitution()
495500

496501
resolvedDrvGoal = worker.makeDerivationGoal(
497502
pathResolved, wantedOutputs, buildMode);
498-
addWaitee(resolvedDrvGoal);
503+
{
504+
Goals waitees{resolvedDrvGoal};
505+
co_await await(std::move(waitees));
506+
}
499507

500-
co_await Suspend{};
501508
co_return resolvedFinished();
502509
}
503510

@@ -536,8 +543,7 @@ Goal::Co DerivationGoal::gaveUpOnSubstitution()
536543
/* Okay, try to build. Note that here we don't wait for a build
537544
slot to become available, since we don't need one if there is a
538545
build hook. */
539-
worker.wakeUp(shared_from_this());
540-
co_await Suspend{};
546+
co_await yield();
541547
co_return tryToBuild();
542548
}
543549

@@ -602,8 +608,7 @@ Goal::Co DerivationGoal::tryToBuild()
602608
/* Wait then try locking again, repeat until success (returned
603609
boolean is true). */
604610
do {
605-
worker.waitForAWhile(shared_from_this());
606-
co_await Suspend{};
611+
co_await waitForAWhile();
607612
} while (!outputLocks.lockPaths(lockFiles, "", false));
608613
}
609614

@@ -667,8 +672,7 @@ Goal::Co DerivationGoal::tryToBuild()
667672

668673
actLock.reset();
669674

670-
worker.wakeUp(shared_from_this());
671-
co_await Suspend{};
675+
co_await yield();
672676
co_return tryLocalBuild();
673677
}
674678

@@ -719,6 +723,8 @@ Goal::Co DerivationGoal::repairClosure()
719723
outputsToDrv.insert_or_assign(*j.second, i);
720724
}
721725

726+
Goals waitees;
727+
722728
/* Check each path (slow!). */
723729
for (auto & i : outputClosure) {
724730
if (worker.pathContentsGood(i)) continue;
@@ -727,27 +733,25 @@ Goal::Co DerivationGoal::repairClosure()
727733
worker.store.printStorePath(i), worker.store.printStorePath(drvPath));
728734
auto drvPath2 = outputsToDrv.find(i);
729735
if (drvPath2 == outputsToDrv.end())
730-
addWaitee(upcast_goal(worker.makePathSubstitutionGoal(i, Repair)));
736+
waitees.insert(upcast_goal(worker.makePathSubstitutionGoal(i, Repair)));
731737
else
732-
addWaitee(worker.makeGoal(
738+
waitees.insert(worker.makeGoal(
733739
DerivedPath::Built {
734740
.drvPath = makeConstantStorePathRef(drvPath2->second),
735741
.outputs = OutputsSpec::All { },
736742
},
737743
bmRepair));
738744
}
739745

740-
if (waitees.empty()) {
741-
co_return done(BuildResult::AlreadyValid, assertPathValidity());
742-
} else {
743-
co_await Suspend{};
746+
co_await await(std::move(waitees));
744747

748+
if (!waitees.empty()) {
745749
trace("closure repaired");
746750
if (nrFailed > 0)
747751
throw Error("some paths in the output closure of derivation '%s' could not be repaired",
748752
worker.store.printStorePath(drvPath));
749-
co_return done(BuildResult::AlreadyValid, assertPathValidity());
750753
}
754+
co_return done(BuildResult::AlreadyValid, assertPathValidity());
751755
}
752756

753757

‎src/libstore/build/drv-output-substitution-goal.cc‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,8 @@ Goal::Co DrvOutputSubstitutionGoal::init()
8787

8888
bool failed = false;
8989

90+
Goals waitees;
91+
9092
for (const auto & [depId, depPath] : outputInfo->dependentRealisations) {
9193
if (depId != id) {
9294
if (auto localOutputInfo = worker.store.queryRealisation(depId);
@@ -103,13 +105,13 @@ Goal::Co DrvOutputSubstitutionGoal::init()
103105
failed = true;
104106
break;
105107
}
106-
addWaitee(worker.makeDrvOutputSubstitutionGoal(depId));
108+
waitees.insert(worker.makeDrvOutputSubstitutionGoal(depId));
107109
}
108110
}
109111

110112
if (failed) continue;
111113

112-
co_return realisationFetched(outputInfo, sub);
114+
co_return realisationFetched(std::move(waitees), outputInfo, sub);
113115
}
114116

115117
/* None left. Terminate this goal and let someone else deal
@@ -127,10 +129,10 @@ Goal::Co DrvOutputSubstitutionGoal::init()
127129
co_return amDone(substituterFailed ? ecFailed : ecNoSubstituters);
128130
}
129131

130-
Goal::Co DrvOutputSubstitutionGoal::realisationFetched(std::shared_ptr<const Realisation> outputInfo, nix::ref<nix::Store> sub) {
131-
addWaitee(worker.makePathSubstitutionGoal(outputInfo->outPath));
132+
Goal::Co DrvOutputSubstitutionGoal::realisationFetched(Goals waitees, std::shared_ptr<const Realisation> outputInfo, nix::ref<nix::Store> sub) {
133+
waitees.insert(worker.makePathSubstitutionGoal(outputInfo->outPath));
132134

133-
if (!waitees.empty()) co_await Suspend{};
135+
co_await await(std::move(waitees));
134136

135137
trace("output path substituted");
136138

‎src/libstore/build/drv-output-substitution-goal.hh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ public:
3434
GoalState state;
3535

3636
Co init() override;
37-
Co realisationFetched(std::shared_ptr<const Realisation> outputInfo, nix::ref<nix::Store> sub);
37+
Co realisationFetched(Goals waitees, std::shared_ptr<const Realisation> outputInfo, nix::ref<nix::Store> sub);
3838

3939
void timedOut(Error && ex) override { unreachable(); };
4040

‎src/libstore/build/goal.cc‎

Lines changed: 52 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -132,38 +132,18 @@ void addToWeakGoals(WeakGoals & goals, GoalPtr p)
132132
goals.insert(p);
133133
}
134134

135-
136-
void Goal::addWaitee(GoalPtr waitee)
137-
{
138-
waitees.insert(waitee);
139-
addToWeakGoals(waitee->waiters, shared_from_this());
140-
}
141-
142-
143-
void Goal::waiteeDone(GoalPtr waitee, ExitCode result)
135+
Co Goal::await(Goals new_waitees)
144136
{
145-
assert(waitees.count(waitee));
146-
waitees.erase(waitee);
147-
148-
trace(fmt("waitee '%s' done; %d left", waitee->name, waitees.size()));
149-
150-
if (result == ecFailed || result == ecNoSubstituters || result == ecIncompleteClosure) ++nrFailed;
151-
152-
if (result == ecNoSubstituters) ++nrNoSubstituters;
153-
154-
if (result == ecIncompleteClosure) ++nrIncompleteClosure;
155-
156-
if (waitees.empty() || (result == ecFailed && !settings.keepGoing)) {
157-
158-
/* If we failed and keepGoing is not set, we remove all
159-
remaining waitees. */
160-
for (auto & goal : waitees) {
161-
goal->waiters.extract(shared_from_this());
137+
assert(waitees.empty());
138+
if (!new_waitees.empty()) {
139+
waitees = std::move(new_waitees);
140+
for (auto waitee : waitees) {
141+
addToWeakGoals(waitee->waiters, shared_from_this());
162142
}
163-
waitees.clear();
164-
165-
worker.wakeUp(shared_from_this());
143+
co_await Suspend{};
144+
assert(waitees.empty());
166145
}
146+
co_return Return{};
167147
}
168148

169149
Goal::Done Goal::amDone(ExitCode result, std::optional<Error> ex)
@@ -183,7 +163,32 @@ Goal::Done Goal::amDone(ExitCode result, std::optional<Error> ex)
183163

184164
for (auto & i : waiters) {
185165
GoalPtr goal = i.lock();
186-
if (goal) goal->waiteeDone(shared_from_this(), result);
166+
if (goal) {
167+
auto me = shared_from_this();
168+
assert(goal->waitees.count(me));
169+
goal->waitees.erase(me);
170+
171+
goal->trace(fmt("waitee '%s' done; %d left", name, goal->waitees.size()));
172+
173+
if (result == ecFailed || result == ecNoSubstituters || result == ecIncompleteClosure) ++goal->nrFailed;
174+
175+
if (result == ecNoSubstituters) ++goal->nrNoSubstituters;
176+
177+
if (result == ecIncompleteClosure) ++goal->nrIncompleteClosure;
178+
179+
if (goal->waitees.empty()) {
180+
worker.wakeUp(goal);
181+
} else if (result == ecFailed && !settings.keepGoing) {
182+
/* If we failed and keepGoing is not set, we remove all
183+
remaining waitees. */
184+
for (auto & g : goal->waitees) {
185+
g->waiters.extract(goal);
186+
}
187+
goal->waitees.clear();
188+
189+
worker.wakeUp(goal);
190+
}
191+
}
187192
}
188193
waiters.clear();
189194
worker.removeGoal(shared_from_this());
@@ -215,5 +220,22 @@ void Goal::work()
215220
assert(top_co || exitCode != ecBusy);
216221
}
217222

223+
Goal::Co Goal::yield() {
224+
worker.wakeUp(shared_from_this());
225+
co_await Suspend{};
226+
co_return Return{};
227+
}
228+
229+
Goal::Co Goal::waitForAWhile() {
230+
worker.waitForAWhile(shared_from_this());
231+
co_await Suspend{};
232+
co_return Return{};
233+
}
234+
235+
Goal::Co Goal::waitForBuildSlot() {
236+
worker.waitForBuildSlot(shared_from_this());
237+
co_await Suspend{};
238+
co_return Return{};
239+
}
218240

219241
}

‎src/libstore/build/goal.hh‎

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -54,18 +54,20 @@ enum struct JobCategory {
5454

5555
struct Goal : public std::enable_shared_from_this<Goal>
5656
{
57+
private:
58+
/**
59+
* Goals that this goal is waiting for.
60+
*/
61+
Goals waitees;
62+
63+
public:
5764
typedef enum {ecBusy, ecSuccess, ecFailed, ecNoSubstituters, ecIncompleteClosure} ExitCode;
5865

5966
/**
6067
* Backlink to the worker.
6168
*/
6269
Worker & worker;
6370

64-
/**
65-
* Goals that this goal is waiting for.
66-
*/
67-
Goals waitees;
68-
6971
/**
7072
* Goals waiting for this one to finish. Must use weak pointers
7173
* here to prevent cycles.
@@ -104,8 +106,8 @@ protected:
104106
* Build result.
105107
*/
106108
BuildResult buildResult;
107-
public:
108109

110+
public:
109111
/**
110112
* Suspend our goal and wait until we get `work`-ed again.
111113
* `co_await`-able by @ref Co.
@@ -332,6 +334,7 @@ public:
332334
std::suspend_always await_transform(Suspend) { return {}; };
333335
};
334336

337+
protected:
335338
/**
336339
* The coroutine being currently executed.
337340
* MUST be updated when switching the coroutine being executed.
@@ -359,6 +362,7 @@ public:
359362
*/
360363
Done amDone(ExitCode result, std::optional<Error> ex = {});
361364

365+
public:
362366
virtual void cleanup() { }
363367

364368
/**
@@ -394,10 +398,6 @@ public:
394398

395399
void work();
396400

397-
void addWaitee(GoalPtr waitee);
398-
399-
void waiteeDone(GoalPtr waitee, ExitCode result);
400-
401401
virtual void handleChildOutput(Descriptor fd, std::string_view data)
402402
{
403403
unreachable();
@@ -429,6 +429,13 @@ public:
429429
* @see JobCategory
430430
*/
431431
virtual JobCategory jobCategory() const = 0;
432+
433+
protected:
434+
Co await(Goals waitees);
435+
436+
Co waitForAWhile();
437+
Co waitForBuildSlot();
438+
Co yield();
432439
};
433440

434441
void addToWeakGoals(WeakGoals & goals, GoalPtr p);

0 commit comments

Comments
 (0)