libstore: use notifications for stats counters

updating statistics *immediately* when any counter changes declutters
things somewhat and makes useful status reports less dependent on the
current worker main loop. using callbacks will make it easier to move
the worker loop into kj entirely, using only promises for scheduling.

Change-Id: I695dfa83111b1ec09b1a54cff268f3c1d7743ed6

10 files changed, 170 insertions, 43 deletions

diff --git a/src/libstore/build/derivation-goal.cc b/src/libstore/build/derivation-goal.cc
index 1dda1b1b4..c288003b4 100644
--- a/src/libstore/build/derivation-goal.cc
+++ b/src/libstore/build/derivation-goal.cc
@@ -71,7 +71,7 @@ DerivationGoal::DerivationGoal(const StorePath & drvPath,
         DerivedPath::Built { makeConstantStorePathRef(drvPath), wantedOutputs }.to_string(worker.store));
     trace("created");
 
-    mcExpectedBuilds = std::make_unique<MaintainCount<uint64_t>>(worker.expectedBuilds);
+    mcExpectedBuilds = worker.expectedBuilds.addTemporarily(1);
 }
 
 
@@ -91,7 +91,7 @@ DerivationGoal::DerivationGoal(const StorePath & drvPath, const BasicDerivation
         DerivedPath::Built { makeConstantStorePathRef(drvPath), drv.outputNames() }.to_string(worker.store));
     trace("created");
 
-    mcExpectedBuilds = std::make_unique<MaintainCount<uint64_t>>(worker.expectedBuilds);
+    mcExpectedBuilds = worker.expectedBuilds.addTemporarily(1);
 
     /* Prevent the .chroot directory from being
        garbage-collected. (See isActiveTempFile() in gc.cc.) */
@@ -662,7 +662,7 @@ void DerivationGoal::started()
     if (hook) msg += fmt(" on '%s'", machineName);
     act = std::make_unique<Activity>(*logger, lvlInfo, actBuild, msg,
         Logger::Fields{worker.store.printStorePath(drvPath), hook ? machineName : "", 1, 1});
-    mcRunningBuilds = std::make_unique<MaintainCount<uint64_t>>(worker.runningBuilds);
+    mcRunningBuilds = worker.runningBuilds.addTemporarily(1);
 }
 
 Goal::WorkResult DerivationGoal::tryToBuild(bool inBuildSlot)
diff --git a/src/libstore/build/derivation-goal.hh b/src/libstore/build/derivation-goal.hh
index 77f9fef4b..bf4a3da93 100644
--- a/src/libstore/build/derivation-goal.hh
+++ b/src/libstore/build/derivation-goal.hh
@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "notifying-counter.hh"
 #include "parsed-derivations.hh"
 #include "lock.hh"
 #include "outputs-spec.hh"
@@ -217,7 +218,7 @@ struct DerivationGoal : public Goal
 
     BuildMode buildMode;
 
-    std::unique_ptr<MaintainCount<uint64_t>> mcExpectedBuilds, mcRunningBuilds;
+    NotifyingCounter<uint64_t>::Bump mcExpectedBuilds, mcRunningBuilds;
 
     std::unique_ptr<Activity> act;
 
diff --git a/src/libstore/build/drv-output-substitution-goal.cc b/src/libstore/build/drv-output-substitution-goal.cc
index 9acff14b0..369b2dd90 100644
--- a/src/libstore/build/drv-output-substitution-goal.cc
+++ b/src/libstore/build/drv-output-substitution-goal.cc
@@ -42,8 +42,7 @@ Goal::WorkResult DrvOutputSubstitutionGoal::tryNext(bool inBuildSlot)
         return WaitForSlot{};
     }
 
-    maintainRunningSubstitutions =
-        std::make_unique<MaintainCount<uint64_t>>(worker.runningSubstitutions);
+    maintainRunningSubstitutions = worker.runningSubstitutions.addTemporarily(1);
 
     if (subs.size() == 0) {
         /* None left.  Terminate this goal and let someone else deal
diff --git a/src/libstore/build/drv-output-substitution-goal.hh b/src/libstore/build/drv-output-substitution-goal.hh
index b48c4670b..8de4d45dd 100644
--- a/src/libstore/build/drv-output-substitution-goal.hh
+++ b/src/libstore/build/drv-output-substitution-goal.hh
@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "notifying-counter.hh"
 #include "store-api.hh"
 #include "goal.hh"
 #include "realisation.hh"
@@ -40,7 +41,7 @@ class DrvOutputSubstitutionGoal : public Goal {
      */
     std::shared_ptr<Store> sub;
 
-    std::unique_ptr<MaintainCount<uint64_t>> maintainRunningSubstitutions;
+    NotifyingCounter<uint64_t>::Bump maintainRunningSubstitutions;
 
     struct DownloadState
     {
diff --git a/src/libstore/build/substitution-goal.cc b/src/libstore/build/substitution-goal.cc
index 1d24938e5..a798cbde2 100644
--- a/src/libstore/build/substitution-goal.cc
+++ b/src/libstore/build/substitution-goal.cc
@@ -21,7 +21,7 @@ PathSubstitutionGoal::PathSubstitutionGoal(
     state = &PathSubstitutionGoal::init;
     name = fmt("substitution of '%s'", worker.store.printStorePath(this->storePath));
     trace("created");
-    maintainExpectedSubstitutions = std::make_unique<MaintainCount<uint64_t>>(worker.expectedSubstitutions);
+    maintainExpectedSubstitutions = worker.expectedSubstitutions.addTemporarily(1);
 }
 
 
@@ -139,11 +139,11 @@ Goal::WorkResult PathSubstitutionGoal::tryNext(bool inBuildSlot)
     /* Update the total expected download size. */
     auto narInfo = std::dynamic_pointer_cast<const NarInfo>(info);
 
-    maintainExpectedNar = std::make_unique<MaintainCount<uint64_t>>(worker.expectedNarSize, info->narSize);
+    maintainExpectedNar = worker.expectedNarSize.addTemporarily(info->narSize);
 
     maintainExpectedDownload =
         narInfo && narInfo->fileSize
-        ? std::make_unique<MaintainCount<uint64_t>>(worker.expectedDownloadSize, narInfo->fileSize)
+        ? worker.expectedDownloadSize.addTemporarily(narInfo->fileSize)
         : nullptr;
 
     /* Bail out early if this substituter lacks a valid
@@ -200,7 +200,7 @@ Goal::WorkResult PathSubstitutionGoal::tryToRun(bool inBuildSlot)
         return WaitForSlot{};
     }
 
-    maintainRunningSubstitutions = std::make_unique<MaintainCount<uint64_t>>(worker.runningSubstitutions);
+    maintainRunningSubstitutions = worker.runningSubstitutions.addTemporarily(1);
 
     outPipe.create();
 
@@ -268,13 +268,10 @@ Goal::WorkResult PathSubstitutionGoal::finished(bool inBuildSlot)
     maintainExpectedSubstitutions.reset();
     worker.doneSubstitutions++;
 
-    if (maintainExpectedDownload) {
-        auto fileSize = maintainExpectedDownload->delta;
-        maintainExpectedDownload.reset();
-        worker.doneDownloadSize += fileSize;
-    }
+    worker.doneDownloadSize += maintainExpectedDownload.delta();
+    maintainExpectedDownload.reset();
 
-    worker.doneNarSize += maintainExpectedNar->delta;
+    worker.doneNarSize += maintainExpectedNar.delta();
     maintainExpectedNar.reset();
 
     return done(ecSuccess, BuildResult::Substituted);
diff --git a/src/libstore/build/substitution-goal.hh b/src/libstore/build/substitution-goal.hh
index 5d58b34a0..9c7e6f470 100644
--- a/src/libstore/build/substitution-goal.hh
+++ b/src/libstore/build/substitution-goal.hh
@@ -2,6 +2,7 @@
 ///@file
 
 #include "lock.hh"
+#include "notifying-counter.hh"
 #include "store-api.hh"
 #include "goal.hh"
 
@@ -63,7 +64,7 @@ struct PathSubstitutionGoal : public Goal
      */
     Path destPath;
 
-    std::unique_ptr<MaintainCount<uint64_t>> maintainExpectedSubstitutions,
+    NotifyingCounter<uint64_t>::Bump maintainExpectedSubstitutions,
         maintainRunningSubstitutions, maintainExpectedNar, maintainExpectedDownload;
 
     typedef WorkResult (PathSubstitutionGoal::*GoalState)(bool inBuildSlot);
diff --git a/src/libstore/build/worker.cc b/src/libstore/build/worker.cc
index 7336ad50f..24c700396 100644
--- a/src/libstore/build/worker.cc
+++ b/src/libstore/build/worker.cc
@@ -320,6 +320,27 @@ void Worker::waitForAWhile(GoalPtr goal)
 }
 
 
+void Worker::updateStatistics()
+{
+    // only update progress info while running. this notably excludes updating
+    // progress info while destroying, which causes the progress bar to assert
+    if (running && statisticsOutdated) {
+        actDerivations.progress(
+            doneBuilds, expectedBuilds + doneBuilds, runningBuilds, failedBuilds
+        );
+        actSubstitutions.progress(
+            doneSubstitutions,
+            expectedSubstitutions + doneSubstitutions,
+            runningSubstitutions,
+            failedSubstitutions
+        );
+        act.setExpected(actFileTransfer, expectedDownloadSize + doneDownloadSize);
+        act.setExpected(actCopyPath, expectedNarSize + doneNarSize);
+
+        statisticsOutdated = false;
+    }
+}
+
 Goals Worker::run(std::function<Goals (GoalFactory &)> req)
 {
     auto _topGoals = req(goalFactory());
@@ -329,6 +350,8 @@ Goals Worker::run(std::function<Goals (GoalFactory &)> req)
     running = true;
     Finally const _stop([&] { running = false; });
 
+    updateStatistics();
+
     for (auto & i : _topGoals) {
         topGoals.insert(i);
         if (auto goal = dynamic_cast<DerivationGoal *>(i.get())) {
@@ -373,18 +396,7 @@ Goals Worker::run(std::function<Goals (GoalFactory &)> req)
                     ? nrSubstitutions < std::max(1U, (unsigned int) settings.maxSubstitutionJobs)
                     : nrLocalBuilds < settings.maxBuildJobs;
                 handleWorkResult(goal, goal->work(inSlot));
-
-                actDerivations.progress(
-                    doneBuilds, expectedBuilds + doneBuilds, runningBuilds, failedBuilds
-                );
-                actSubstitutions.progress(
-                    doneSubstitutions,
-                    expectedSubstitutions + doneSubstitutions,
-                    runningSubstitutions,
-                    failedSubstitutions
-                );
-                act.setExpected(actFileTransfer, expectedDownloadSize + doneDownloadSize);
-                act.setExpected(actCopyPath, expectedNarSize + doneNarSize);
+                updateStatistics();
 
                 if (topGoals.empty()) break; // stuff may have been cancelled
             }
diff --git a/src/libstore/build/worker.hh b/src/libstore/build/worker.hh
index 3fbf457fe..9a6ed8449 100644
--- a/src/libstore/build/worker.hh
+++ b/src/libstore/build/worker.hh
@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "notifying-counter.hh"
 #include "types.hh"
 #include "lock.hh"
 #include "store-api.hh"
@@ -213,6 +214,21 @@ private:
     void childStarted(GoalPtr goal, const std::set<int> & fds,
         bool inBuildSlot);
 
+    /**
+      * Pass current stats counters to the logger for progress bar updates.
+      */
+    void updateStatistics();
+
+    bool statisticsOutdated = true;
+
+    /**
+      * Mark statistics as outdated, such that `updateStatistics` will be called.
+      */
+    void updateStatisticsLater()
+    {
+        statisticsOutdated = true;
+    }
+
 public:
 
     const Activity act;
@@ -234,19 +250,19 @@ public:
 
     HookState hook;
 
-    uint64_t expectedBuilds = 0;
-    uint64_t doneBuilds = 0;
-    uint64_t failedBuilds = 0;
-    uint64_t runningBuilds = 0;
-
-    uint64_t expectedSubstitutions = 0;
-    uint64_t doneSubstitutions = 0;
-    uint64_t failedSubstitutions = 0;
-    uint64_t runningSubstitutions = 0;
-    uint64_t expectedDownloadSize = 0;
-    uint64_t doneDownloadSize = 0;
-    uint64_t expectedNarSize = 0;
-    uint64_t doneNarSize = 0;
+    NotifyingCounter<uint64_t> expectedBuilds{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> doneBuilds{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> failedBuilds{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> runningBuilds{[this] { updateStatisticsLater(); }};
+
+    NotifyingCounter<uint64_t> expectedSubstitutions{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> doneSubstitutions{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> failedSubstitutions{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> runningSubstitutions{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> expectedDownloadSize{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> doneDownloadSize{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> expectedNarSize{[this] { updateStatisticsLater(); }};
+    NotifyingCounter<uint64_t> doneNarSize{[this] { updateStatisticsLater(); }};
 
     Worker(Store & store, Store & evalStore);
     ~Worker();
diff --git a/src/libutil/meson.build b/src/libutil/meson.build
index 6566f7f46..1ac31c7eb 100644
--- a/src/libutil/meson.build
+++ b/src/libutil/meson.build
@@ -95,6 +95,7 @@ libutil_headers = files(
   'monitor-fd.hh',
   'mount.hh',
   'namespaces.hh',
+  'notifying-counter.hh',
   'pool.hh',
   'position.hh',
   'print-elided.hh',
diff --git a/src/libutil/notifying-counter.hh b/src/libutil/notifying-counter.hh
new file mode 100644
index 000000000..dc58aac91
--- /dev/null
+++ b/src/libutil/notifying-counter.hh
@@ -0,0 +1,99 @@
+#pragma once
+/// @file
+
+#include <cassert>
+#include <functional>
+#include <memory>
+
+namespace nix {
+
+template<std::integral T>
+class NotifyingCounter
+{
+private:
+    T counter;
+    std::function<void()> notify;
+
+public:
+    class Bump
+    {
+        friend class NotifyingCounter;
+
+        struct SubOnFree
+        {
+            T delta;
+
+            void operator()(NotifyingCounter * c) const
+            {
+                c->add(-delta);
+            }
+        };
+
+        // lightly misuse unique_ptr to get RAII types with destructor callbacks
+        std::unique_ptr<NotifyingCounter<T>, SubOnFree> at;
+
+        Bump(NotifyingCounter<T> & at, T delta) : at(&at, {delta}) {}
+
+    public:
+        Bump() = default;
+        Bump(decltype(nullptr)) {}
+
+        T delta() const
+        {
+            return at ? at.get_deleter().delta : 0;
+        }
+
+        void reset()
+        {
+            at.reset();
+        }
+    };
+
+    explicit NotifyingCounter(std::function<void()> notify, T initial = 0)
+        : counter(initial)
+        , notify(std::move(notify))
+    {
+        assert(this->notify);
+    }
+
+    // bumps hold pointers to this, so we should neither copy nor move.
+    NotifyingCounter(const NotifyingCounter &) = delete;
+    NotifyingCounter & operator=(const NotifyingCounter &) = delete;
+    NotifyingCounter(NotifyingCounter &&) = delete;
+    NotifyingCounter & operator=(NotifyingCounter &&) = delete;
+
+    T get() const
+    {
+        return counter;
+    }
+
+    operator T() const
+    {
+        return counter;
+    }
+
+    void add(T delta)
+    {
+        counter += delta;
+        notify();
+    }
+
+    NotifyingCounter & operator+=(T delta)
+    {
+        add(delta);
+        return *this;
+    }
+
+    NotifyingCounter & operator++(int)
+    {
+        return *this += 1;
+    }
+
+    Bump addTemporarily(T delta)
+    {
+        add(delta);
+        return Bump{*this, delta};
+    }
+};
+
+}