1 files changed, 29 insertions, 41 deletions

diff --git a/src/libstore/build/drv-output-substitution-goal.cc b/src/libstore/build/drv-output-substitution-goal.cc
index 7986123cc..9ffa33d7b 100644
--- a/src/libstore/build/drv-output-substitution-goal.cc
+++ b/src/libstore/build/drv-output-substitution-goal.cc
@@ -4,6 +4,9 @@
 #include "worker.hh"
 #include "substitution-goal.hh"
 #include "signals.hh"
+#include <kj/array.h>
+#include <kj/async.h>
+#include <kj/vector.h>
 
 namespace nix {
 
@@ -16,33 +19,32 @@ DrvOutputSubstitutionGoal::DrvOutputSubstitutionGoal(
     : Goal(worker, isDependency)
     , id(id)
 {
-    state = &DrvOutputSubstitutionGoal::init;
     name = fmt("substitution of '%s'", id.to_string());
     trace("created");
 }
 
 
-kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::init(bool inBuildSlot) noexcept
+kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::work() noexcept
 try {
     trace("init");
 
     /* If the derivation already exists, we’re done */
     if (worker.store.queryRealisation(id)) {
-        return {Finished{ecSuccess, std::move(buildResult)}};
+        co_return Finished{ecSuccess, std::move(buildResult)};
     }
 
     subs = settings.useSubstitutes ? getDefaultSubstituters() : std::list<ref<Store>>();
-    return tryNext(inBuildSlot);
+    co_return co_await tryNext();
 } catch (...) {
-    return {std::current_exception()};
+    co_return result::failure(std::current_exception());
 }
 
-kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::tryNext(bool inBuildSlot) noexcept
+kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::tryNext() noexcept
 try {
     trace("trying next substituter");
 
-    if (!inBuildSlot) {
-        return {WaitForSlot{}};
+    if (!slotToken.valid()) {
+        slotToken = co_await worker.substitutions.acquire();
     }
 
     maintainRunningSubstitutions = worker.runningSubstitutions.addTemporarily(1);
@@ -59,7 +61,7 @@ try {
         /* Hack: don't indicate failure if there were no substituters.
            In that case the calling derivation should just do a
            build. */
-        return {Finished{substituterFailed ? ecFailed : ecNoSubstituters, std::move(buildResult)}};
+        co_return Finished{substituterFailed ? ecFailed : ecNoSubstituters, std::move(buildResult)};
     }
 
     sub = subs.front();
@@ -69,25 +71,26 @@ try {
        some other error occurs), so it must not touch `this`. So put
        the shared state in a separate refcounted object. */
     downloadState = std::make_shared<DownloadState>();
-    downloadState->outPipe.create();
+    auto pipe = kj::newPromiseAndCrossThreadFulfiller<void>();
+    downloadState->outPipe = kj::mv(pipe.fulfiller);
 
     downloadState->result =
         std::async(std::launch::async, [downloadState{downloadState}, id{id}, sub{sub}] {
+            Finally updateStats([&]() { downloadState->outPipe->fulfill(); });
             ReceiveInterrupts receiveInterrupts;
-            Finally updateStats([&]() { downloadState->outPipe.writeSide.close(); });
             return sub->queryRealisation(id);
         });
 
-    state = &DrvOutputSubstitutionGoal::realisationFetched;
-    return {WaitForWorld{{downloadState->outPipe.readSide.get()}, true}};
+    co_await pipe.promise;
+    co_return co_await realisationFetched();
 } catch (...) {
-    return {std::current_exception()};
+    co_return result::failure(std::current_exception());
 }
 
-kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::realisationFetched(bool inBuildSlot) noexcept
+kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::realisationFetched() noexcept
 try {
-    worker.childTerminated(this);
     maintainRunningSubstitutions.reset();
+    slotToken = {};
 
     try {
         outputInfo = downloadState->result.get();
@@ -97,10 +100,10 @@ try {
     }
 
     if (!outputInfo) {
-        return tryNext(inBuildSlot);
+        co_return co_await tryNext();
     }
 
-    WaitForGoals result;
+    kj::Vector<std::pair<GoalPtr, kj::Promise<void>>> dependencies;
     for (const auto & [depId, depPath] : outputInfo->dependentRealisations) {
         if (depId != id) {
             if (auto localOutputInfo = worker.store.queryRealisation(depId);
@@ -114,25 +117,23 @@ try {
                     worker.store.printStorePath(localOutputInfo->outPath),
                     worker.store.printStorePath(depPath)
                 );
-                return tryNext(inBuildSlot);
+                co_return co_await tryNext();
             }
-            result.goals.insert(worker.goalFactory().makeDrvOutputSubstitutionGoal(depId));
+            dependencies.add(worker.goalFactory().makeDrvOutputSubstitutionGoal(depId));
         }
     }
 
-    result.goals.insert(worker.goalFactory().makePathSubstitutionGoal(outputInfo->outPath));
+    dependencies.add(worker.goalFactory().makePathSubstitutionGoal(outputInfo->outPath));
 
-    if (result.goals.empty()) {
-        return outPathValid(inBuildSlot);
-    } else {
-        state = &DrvOutputSubstitutionGoal::outPathValid;
-        return {std::move(result)};
+    if (!dependencies.empty()) {
+        (co_await waitForGoals(dependencies.releaseAsArray())).value();
     }
+    co_return co_await outPathValid();
 } catch (...) {
-    return {std::current_exception()};
+    co_return result::failure(std::current_exception());
 }
 
-kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::outPathValid(bool inBuildSlot) noexcept
+kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::outPathValid() noexcept
 try {
     assert(outputInfo);
     trace("output path substituted");
@@ -159,17 +160,4 @@ try {
     return {std::current_exception()};
 }
 
-std::string DrvOutputSubstitutionGoal::key()
-{
-    /* "a$" ensures substitution goals happen before derivation
-       goals. */
-    return "a$" + std::string(id.to_string());
-}
-
-kj::Promise<Result<Goal::WorkResult>> DrvOutputSubstitutionGoal::work(bool inBuildSlot) noexcept
-{
-    return (this->*state)(inBuildSlot);
-}
-
-
 }