libstore/build: fix starvation during substitution

When the destructor of PathSubstitutionGoal is run, this happens in a
sync context and can cause starvation of all ongoing IO w.r.t. to other
substitutions, including our own substitution.

While there's only a decompressor thread per stream, the other side of
the IO runs on the event loop.

In order to fix this, it is sufficient to remove the thread indirection
and inline the async code.

Fixes #1126. Great thanks to horrors' patience.

Co-authored-by: eldritch horrors <pennae@lix.systems>
Change-Id: I3eb37bc37d156f0f5528364e568fdaa2ced58011
Signed-off-by: Raito Bezarius <raito@lix.systems>

lix/libstore/build/derivation-goal.cc

@@ -139,7 +139,14 @@ Goal::WorkResult DerivationGoal::timedOut(Error && ex)
 
 kj::Promise<Result<Goal::WorkResult>> DerivationGoal::workImpl() noexcept
 {
-    KJ_DEFER({ actLock.reset(); });
+    // always clear the slot token, no matter what happens. not doing this
+    // can cause builds to get stuck on exceptions (or other early exits).
+    // ideally we'd use scoped slot tokens instead of keeping them in some
+    // goal member variable, but we cannot do this yet for legacy reasons.
+    KJ_DEFER({
+        actLock.reset();
+        slotToken = {};
+    });
 
     BOOST_OUTCOME_CO_TRY(auto result, co_await (useDerivation ? getDerivation() : haveDerivation()));
     result.storePath = drvPath;

lix/libstore/build/derivation-goal.hh

@@ -246,7 +246,7 @@ struct DerivationGoal : public Goal
 
     WorkResult timedOut(Error && ex);
 
-    kj::Promise<Result<WorkResult>> workImpl() noexcept override;
+    kj::Promise<Result<WorkResult>> workImpl() noexcept override final;
 
     /**
      * Add wanted outputs to an already existing derivation goal.
@@ -307,6 +307,8 @@ struct DerivationGoal : public Goal
     virtual void cleanupPostOutputsRegisteredModeNonCheck();
 
 protected:
+    AsyncSemaphore::Token slotToken;
+
     kj::TimePoint lastChildActivity = kj::minValue;
 
     kj::Promise<Result<WorkResult>> wrapChildHandler(kj::Promise<Result<WorkResult>> handler

lix/libstore/build/goal.cc

@@ -22,12 +22,6 @@ kj::Promise<void> Goal::waitForAWhile()
 
 kj::Promise<Result<Goal::WorkResult>> Goal::work() noexcept
 try {
-    // always clear the slot token, no matter what happens. not doing this
-    // can cause builds to get stuck on exceptions (or other early exist).
-    // ideally we'd use scoped slot tokens instead of keeping them in some
-    // goal member variable, but we cannot do this yet for legacy reasons.
-    KJ_DEFER({ slotToken = {}; });
-
     BOOST_OUTCOME_CO_TRY(auto result, co_await workImpl());
 
     trace("done");

lix/libstore/build/goal.hh

@@ -82,9 +82,6 @@ struct Goal
      */
     std::string name;
 
-protected:
-    AsyncSemaphore::Token slotToken;
-
 public:
     struct [[nodiscard]] WorkResult {
         ExitCode exitCode;

lix/libstore/build/substitution-goal.cc

@@ -27,13 +27,6 @@ PathSubstitutionGoal::PathSubstitutionGoal(
     maintainExpectedSubstitutions = worker.expectedSubstitutions.addTemporarily(1);
 }
 
-
-PathSubstitutionGoal::~PathSubstitutionGoal()
-{
-    cleanup();
-}
-
-
 Goal::WorkResult PathSubstitutionGoal::done(
     ExitCode result,
     BuildResult::Status status,
@@ -76,8 +69,6 @@ kj::Promise<Result<Goal::WorkResult>> PathSubstitutionGoal::tryNext() noexcept
 try {
     trace("trying next substituter");
 
-    cleanup();
-
     if (subs.size() == 0) {
         /* None left.  Terminate this goal and let someone else deal
            with it. */
@@ -206,61 +197,36 @@ kj::Promise<Result<Goal::WorkResult>> PathSubstitutionGoal::tryToRun() noexcept
 try {
     trace("trying to run");
 
-    if (!slotToken.valid()) {
-        slotToken = co_await worker.substitutions.acquire();
-    }
-
-    maintainRunningSubstitutions = worker.runningSubstitutions.addTemporarily(1);
-
-    auto pipe = kj::newPromiseAndCrossThreadFulfiller<void>();
-    outPipe = kj::mv(pipe.fulfiller);
-
-    thr = std::async(std::launch::async, [this]() {
-        AsyncIoRoot aio;
-        /* Wake up the worker loop when we're done. */
-        Finally updateStats([this]() { outPipe->fulfill(); });
-
-        auto & fetchPath = subPath ? *subPath : storePath;
-        try {
-            ReceiveInterrupts receiveInterrupts;
-
-            auto act = logger->startActivity(
-                actSubstitute, Logger::Fields{worker.store.printStorePath(storePath), sub->getUri()}
-            );
-
-            aio.blockOn(copyStorePath(
-                *sub,
-                worker.store,
-                fetchPath,
-                repair,
-                sub->config().isTrusted ? NoCheckSigs : CheckSigs,
-                &act
-            ));
-        } catch (const EndOfFile &) {
-            throw EndOfFile(
-                "NAR for '%s' fetched from '%s' is incomplete",
-                sub->printStorePath(fetchPath),
-                sub->getUri()
-            );
-        }
-    });
-
-    co_await pipe.promise;
-    co_return co_await finished();
-} catch (...) {
-    co_return result::current_exception();
-}
-
-
-kj::Promise<Result<Goal::WorkResult>> PathSubstitutionGoal::finished() noexcept
-try {
-    trace("substitute finished");
-
+    auto & fetchPath = subPath ? *subPath : storePath;
     do {
         try {
-            slotToken = {};
-            thr.get();
-            break;
+            try {
+                AsyncSemaphore::Token slotToken = co_await worker.substitutions.acquire();
+
+                auto act = logger->startActivity(
+                    actSubstitute,
+                    Logger::Fields{worker.store.printStorePath(storePath), sub->getUri()}
+                );
+
+                maintainRunningSubstitutions = worker.runningSubstitutions.addTemporarily(1);
+
+                TRY_AWAIT(copyStorePath(
+                    *sub,
+                    worker.store,
+                    fetchPath,
+                    repair,
+                    sub->config().isTrusted ? NoCheckSigs : CheckSigs,
+                    &act
+                ));
+
+                break;
+            } catch (const EndOfFile &) {
+                throw EndOfFile(
+                    "NAR for '%s' fetched from '%s' is incomplete",
+                    sub->printStorePath(fetchPath),
+                    sub->getUri()
+                );
+            }
         } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions)
             printError("%1%", Uncolored(e.what()));
 
@@ -272,10 +238,20 @@ try {
                 substituterFailed = true;
             }
         }
-        /* Try the next substitute. */
+
+        /* Try the next substitute */
         co_return co_await tryNext();
     } while (false);
 
+    co_return co_await finished();
+} catch (...) {
+    co_return result::current_exception();
+}
+
+kj::Promise<Result<Goal::WorkResult>> PathSubstitutionGoal::finished() noexcept
+try {
+    trace("substitute finished");
+
     worker.markContentsGood(storePath);
 
     printMsg(lvlChatty, "substitution of path '%s' succeeded", worker.store.printStorePath(storePath));
@@ -295,19 +271,4 @@ try {
 } catch (...) {
     co_return result::current_exception();
 }
-
-
-void PathSubstitutionGoal::cleanup()
-{
-    try {
-        if (thr.valid()) {
-            // FIXME: signal worker thread to quit.
-            thr.get();
-        }
-    } catch (...) {
-        ignoreExceptionInDestructor();
-    }
-}
-
-
 }

lix/libstore/build/substitution-goal.hh

@@ -48,11 +48,6 @@ struct PathSubstitutionGoal : public Goal
      */
     kj::Own<kj::CrossThreadPromiseFulfiller<void>> outPipe;
 
-    /**
-     * The substituter thread.
-     */
-    std::future<void> thr;
-
     /**
      * Whether to try to repair a valid path.
      */
@@ -85,7 +80,6 @@ public:
         RepairFlag repair = NoRepair,
         std::optional<ContentAddress> ca = std::nullopt
     );
-    ~PathSubstitutionGoal();
 
     kj::Promise<Result<WorkResult>> workImpl() noexcept override;
 
@@ -97,9 +91,6 @@ public:
     kj::Promise<Result<WorkResult>> tryToRun() noexcept;
     kj::Promise<Result<WorkResult>> finished() noexcept;
 
-    /* Called by destructor, can't be overridden */
-    void cleanup() override final;
-
     JobCategory jobCategory() const override {
         return JobCategory::Substitution;
     };