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.

2 files changed, 29 insertions, 2 deletions

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; }
 };
 
 }