Add saturation metric for executors.

This should make it easier to observe bottlenecks in one of the
underlying executor threads used in the "field writer" SequencedTaskExecutor.

4 files changed, 34 insertions, 2 deletions

diff --git a/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.cpp b/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.cpp
index 7c4dcf434a3..b7079542c06 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.cpp
+++ b/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.cpp
@@ -11,6 +11,7 @@ ExecutorMetrics::update(const vespalib::ExecutorStats &stats)
     rejected.inc(stats.rejectedTasks);
     wakeupCount.inc(stats.wakeupCount);
     util.set(stats.getUtil());
+    saturation.set(stats.get_saturation());
     const auto & qSize = stats.queueSize;
     queueSize.addValueBatch(qSize.average(), qSize.count(), qSize.min(), qSize.max());
 }
@@ -21,6 +22,9 @@ ExecutorMetrics::ExecutorMetrics(const std::string &name, metrics::MetricSet *pa
       rejected("rejected", {}, "Number of rejected tasks", this),
       wakeupCount("wakeups", {}, "Number of times a worker thread has been woken up", this),
       util("utilization", {}, "Ratio of time the worker threads has been active", this),
+      saturation("saturation", {}, "Ratio indicating how saturated the worker threads has been. "
+                                   "For most executors this ratio is equal to utilization, but for others (e.g SequencedTaskExecutor) "
+                                   " a higher saturation than utilization indicates a bottleneck in a subset of the worker threads.", this),
       queueSize("queuesize", {}, "Size of task queue", this)
 {
 }
diff --git a/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.h b/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.h
index af3a27da240..cd6181b108c 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.h
+++ b/searchcore/src/vespa/searchcore/proton/metrics/executor_metrics.h
@@ -15,6 +15,7 @@ struct ExecutorMetrics : metrics::MetricSet
     metrics::LongCountMetric   rejected;
     metrics::LongCountMetric   wakeupCount;
     metrics::DoubleValueMetric util;
+    metrics::DoubleValueMetric saturation;
     metrics::LongAverageMetric queueSize;
 
     void update(const vespalib::ExecutorStats &stats);
diff --git a/vespalib/src/tests/executor/threadstackexecutor_test.cpp b/vespalib/src/tests/executor/threadstackexecutor_test.cpp
index ad412bac3d2..9dd7e37d580 100644
--- a/vespalib/src/tests/executor/threadstackexecutor_test.cpp
+++ b/vespalib/src/tests/executor/threadstackexecutor_test.cpp
@@ -214,6 +214,27 @@ TEST("require that stats can be accumulated") {
     EXPECT_EQUAL(0.41, stats.getUtil());
 }
 
+ExecutorStats make_stats(uint32_t thread_count, double idle) {
+    ExecutorStats stats;
+    stats.setUtil(thread_count, idle);
+    return stats;
+}
+
+TEST("executor stats saturation is the max of the utilization of aggregated executor stats") {
+    ExecutorStats aggr;
+    auto s1 = make_stats(1, 0.9);
+    EXPECT_EQUAL(0.1, s1.getUtil());
+    EXPECT_EQUAL(0.1, s1.get_saturation());
+
+    EXPECT_EQUAL(0.0, aggr.get_saturation());
+    aggr.aggregate(s1);
+    EXPECT_EQUAL(0.1, aggr.get_saturation());
+    aggr.aggregate(make_stats(1, 0.7));
+    EXPECT_EQUAL(0.3, aggr.get_saturation());
+    aggr.aggregate(make_stats(1, 0.8));
+    EXPECT_EQUAL(0.3, aggr.get_saturation());
+}
+
 TEST("Test that utilization is computed") {
     ThreadStackExecutor executor(1);
     std::this_thread::sleep_for(1s);
diff --git a/vespalib/src/vespa/vespalib/util/executor_stats.h b/vespalib/src/vespa/vespalib/util/executor_stats.h
index 577ae933ec2..4177aa07b1d 100644
--- a/vespalib/src/vespa/vespalib/util/executor_stats.h
+++ b/vespalib/src/vespa/vespalib/util/executor_stats.h
@@ -2,8 +2,9 @@
 
 #pragma once
 
-#include <limits>
+#include <algorithm>
 #include <cstdint>
+#include <limits>
 
 namespace vespalib {
 
@@ -57,6 +58,7 @@ class ExecutorStats {
 private:
     size_t   _threadCount;
     double   _absUtil;
+    double   _saturation;
 public:
     using QueueSizeT = AggregatedAverage<size_t>;
     QueueSizeT queueSize;
@@ -67,7 +69,8 @@ public:
     ExecutorStats() : ExecutorStats(QueueSizeT(), 0, 0, 0) {}
     ExecutorStats(QueueSizeT queueSize_in, size_t accepted, size_t rejected, size_t wakeupCount_in)
         : _threadCount(1),
-          _absUtil(1.0),
+          _absUtil(0.0),
+          _saturation(0.0),
           queueSize(queueSize_in),
           acceptedTasks(accepted),
           rejectedTasks(rejected),
@@ -83,14 +86,17 @@ public:
         rejectedTasks += rhs.rejectedTasks;
         wakeupCount += rhs.wakeupCount;
         _absUtil += rhs._absUtil;
+        _saturation = std::max(_saturation, rhs.get_saturation());
     }
     ExecutorStats & setUtil(uint32_t threadCount, double idle) {
         _threadCount = threadCount;
         _absUtil = (1.0 - idle) * threadCount;
+        _saturation = getUtil();
         return *this;
     }
     double getUtil() const { return _absUtil / _threadCount; }
     size_t getThreadCount() const { return _threadCount; }
+    double get_saturation() const { return _saturation; }
 };
 
 }