Merge commit 'f4c869977c' into progress-bar

src/libstore/binary-cache-store.cc

@@ -441,16 +441,25 @@ void BinaryCacheStore::queryRealisationUncached(const DrvOutput & id,
     Callback<std::shared_ptr<const Realisation>> callback) noexcept
 {
     auto outputInfoFilePath = realisationsPrefix + "/" + id.to_string() + ".doi";
-    auto rawOutputInfo = getFile(outputInfoFilePath);
 
-    if (rawOutputInfo) {
+    auto callbackPtr = std::make_shared<decltype(callback)>(std::move(callback));
+
+    Callback<std::shared_ptr<std::string>> newCallback = {
+        [=](std::future<std::shared_ptr<std::string>> fut) {
+            try {
+                auto data = fut.get();
+                if (!data) return (*callbackPtr)(nullptr);
+
                 auto realisation = Realisation::fromJSON(
-            nlohmann::json::parse(*rawOutputInfo), outputInfoFilePath);
-        callback(std::make_shared<const Realisation>(realisation));
-        return;
-    } else {
-        callback(nullptr);
+                    nlohmann::json::parse(*data), outputInfoFilePath);
+                return (*callbackPtr)(std::make_shared<const Realisation>(realisation));
+            } catch (...) {
+                callbackPtr->rethrow();
             }
+        }
+    };
+
+    getFile(outputInfoFilePath, std::move(newCallback));
 }
 
 void BinaryCacheStore::registerDrvOutput(const Realisation& info) {

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

@@ -1,6 +1,8 @@
 #include "drv-output-substitution-goal.hh"
+#include "finally.hh"
 #include "worker.hh"
 #include "substitution-goal.hh"
+#include "callback.hh"
 
 namespace nix {
 
@@ -50,14 +52,42 @@ void DrvOutputSubstitutionGoal::tryNext()
         return;
     }
 
-    auto sub = subs.front();
+    sub = subs.front();
     subs.pop_front();
 
     // FIXME: Make async
-    outputInfo = sub->queryRealisation(id);
+    // outputInfo = sub->queryRealisation(id);
+    outPipe.create();
+    promise = decltype(promise)();
+
+    sub->queryRealisation(
+        id, { [&](std::future<std::shared_ptr<const Realisation>> res) {
+            try {
+                Finally updateStats([this]() { outPipe.writeSide.close(); });
+                promise.set_value(res.get());
+            } catch (...) {
+                promise.set_exception(std::current_exception());
+            }
+        } });
+
+    worker.childStarted(shared_from_this(), {outPipe.readSide.get()}, true, false);
+
+    state = &DrvOutputSubstitutionGoal::realisationFetched;
+}
+
+void DrvOutputSubstitutionGoal::realisationFetched()
+{
+    worker.childTerminated(this);
+
+    try {
+        outputInfo = promise.get_future().get();
+    } catch (std::exception & e) {
+        printError(e.what());
+        substituterFailed = true;
+    }
+
     if (!outputInfo) {
-        tryNext();
-        return;
+        return tryNext();
     }
 
     for (const auto & [depId, depPath] : outputInfo->dependentRealisations) {
@@ -119,4 +149,10 @@ void DrvOutputSubstitutionGoal::work()
     (this->*state)();
 }
 
+void DrvOutputSubstitutionGoal::handleEOF(int fd)
+{
+    if (fd == outPipe.readSide.get()) worker.wakeUp(shared_from_this());
+}
+
+
 }

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

@@ -3,6 +3,8 @@
 #include "store-api.hh"
 #include "goal.hh"
 #include "realisation.hh"
+#include <thread>
+#include <future>
 
 namespace nix {
 
@@ -25,6 +27,13 @@ private:
     /* The remaining substituters. */
     std::list<ref<Store>> subs;
 
+    /* The current substituter. */
+    std::shared_ptr<Store> sub;
+
+    Pipe outPipe;
+    std::thread thr;
+    std::promise<std::shared_ptr<const Realisation>> promise;
+
     /* Whether a substituter failed. */
     bool substituterFailed = false;
 
@@ -36,6 +45,7 @@ public:
 
     void init();
     void tryNext();
+    void realisationFetched();
     void outPathValid();
     void finished();
 
@@ -44,7 +54,7 @@ public:
     string key() override;
 
     void work() override;
-
+    void handleEOF(int fd) override;
 };
 
 }