libstore: limit CA realisation info substitution concurrency

this seems to be an oversight, considering that regular substitutions
are concurrency-limited. while not particularly necessary at present,
once we've removed the `Callback` based interfaces it will be needed.

Change-Id: Ide2d08169fcc24752cbd07a1d33fb8482f7034f5

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

@@ -3,6 +3,7 @@
 #include "worker.hh"
 #include "substitution-goal.hh"
 #include "callback.hh"
+#include "signals.hh"
 
 namespace nix {
 
@@ -38,6 +39,18 @@ void DrvOutputSubstitutionGoal::tryNext()
 {
     trace("trying next substituter");
 
+    /* Make sure that we are allowed to start a substitution.  Note that even
+       if maxSubstitutionJobs == 0, we still allow a substituter to run. This
+       prevents infinite waiting. */
+    if (worker.runningCASubstitutions >= std::max(1U, settings.maxSubstitutionJobs.get())) {
+        worker.waitForBuildSlot(shared_from_this());
+        return;
+    }
+
+    maintainRunningSubstitutions =
+        std::make_unique<MaintainCount<uint64_t>>(worker.runningCASubstitutions);
+    worker.updateProgress();
+
     if (subs.size() == 0) {
         /* None left.  Terminate this goal and let someone else deal
            with it. */
@@ -87,6 +100,7 @@ void DrvOutputSubstitutionGoal::tryNext()
 void DrvOutputSubstitutionGoal::realisationFetched()
 {
     worker.childTerminated(this);
+    maintainRunningSubstitutions.reset();
 
     try {
         outputInfo = downloadState->promise.get_future().get();

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

@@ -41,6 +41,13 @@ class DrvOutputSubstitutionGoal : public Goal {
      */
     std::shared_ptr<Store> sub;
 
+    /**
+     * The substituter thread.
+     */
+    std::thread thr;
+
+    std::unique_ptr<MaintainCount<uint64_t>> maintainRunningSubstitutions;
+
     struct DownloadState
     {
         Pipe outPipe;

src/libstore/build/worker.hh

@@ -166,6 +166,7 @@ public:
     uint64_t doneSubstitutions = 0;
     uint64_t failedSubstitutions = 0;
     uint64_t runningSubstitutions = 0;
+    uint64_t runningCASubstitutions = 0;
     uint64_t expectedDownloadSize = 0;
     uint64_t doneDownloadSize = 0;
     uint64_t expectedNarSize = 0;