Merge pull request #20937 from vespa-engine/havardpe/tag-cpu-usage

tag match/summary engines with READ cpu usage

6 files changed, 59 insertions, 17 deletions

diff --git a/vespalib/src/vespa/vespalib/util/cpu_usage.cpp b/vespalib/src/vespa/vespalib/util/cpu_usage.cpp
index 5307239118d..97dbedad66f 100644
--- a/vespalib/src/vespa/vespalib/util/cpu_usage.cpp
+++ b/vespalib/src/vespa/vespalib/util/cpu_usage.cpp
@@ -263,4 +263,29 @@ CpuUsage::sample()
     return self().sample_or_wait();
 }
 
+Runnable::init_fun_t
+CpuUsage::wrap(Runnable::init_fun_t init, Category cat)
+{
+    return [init,cat](Runnable &target) {
+        auto my_usage = CpuUsage::use(cat);
+        return init(target);
+    };
+}
+
+Executor::Task::UP
+CpuUsage::wrap(Executor::Task::UP task, Category cat)
+{
+    struct CpuTask : Executor::Task {
+        UP task;
+        Category cat;
+        CpuTask(UP task_in, Category cat_in)
+          : task(std::move(task_in)), cat(cat_in) {}
+        void run() override {
+            auto my_usage = CpuUsage::use(cat);
+            task->run();
+        }
+    };
+    return std::make_unique<CpuTask>(std::move(task), cat);
+}
+
 } // namespace
diff --git a/vespalib/src/vespa/vespalib/util/cpu_usage.h b/vespalib/src/vespa/vespalib/util/cpu_usage.h
index 87e0a289e87..6fd1a370887 100644
--- a/vespalib/src/vespa/vespalib/util/cpu_usage.h
+++ b/vespalib/src/vespa/vespalib/util/cpu_usage.h
@@ -1,5 +1,7 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
+#include "runnable.h"
+#include "executor.h"
 #include "spin_lock.h"
 #include <vespa/vespalib/util/time.h>
 #include <vespa/vespalib/stllike/string.h>
@@ -178,6 +180,8 @@ private:
 public:
     static MyUsage use(Category cat) { return MyUsage(cat); }
     static TimedSample sample();
+    static Runnable::init_fun_t wrap(Runnable::init_fun_t init, Category cat);
+    static Executor::Task::UP wrap(Executor::Task::UP task, Category cat);
 };
 
 /**
diff --git a/vespalib/src/vespa/vespalib/util/runnable.cpp b/vespalib/src/vespa/vespalib/util/runnable.cpp
index c67c4696d88..3b75efe93f6 100644
--- a/vespalib/src/vespa/vespalib/util/runnable.cpp
+++ b/vespalib/src/vespa/vespalib/util/runnable.cpp
@@ -4,4 +4,11 @@
 
 namespace vespalib {
 
+int
+Runnable::default_init_function(Runnable &target)
+{
+    target.run();
+    return 1;
+}
+
 } // namespace vespalib
diff --git a/vespalib/src/vespa/vespalib/util/runnable.h b/vespalib/src/vespa/vespalib/util/runnable.h
index 43144ebc2cd..bc89a9decac 100644
--- a/vespalib/src/vespa/vespalib/util/runnable.h
+++ b/vespalib/src/vespa/vespalib/util/runnable.h
@@ -23,6 +23,7 @@ namespace vespalib {
 struct Runnable {
     using UP = std::unique_ptr<Runnable>;
     using init_fun_t = std::function<int(Runnable&)>;
+    static int default_init_function(Runnable &target);
 
     /**
      * Entry point called by the running thread
@@ -36,4 +37,3 @@ struct Runnable {
 };
 
 } // namespace vespalib
-
diff --git a/vespalib/src/vespa/vespalib/util/simple_thread_bundle.cpp b/vespalib/src/vespa/vespalib/util/simple_thread_bundle.cpp
index ab83d4e05fd..99ab298864f 100644
--- a/vespalib/src/vespa/vespalib/util/simple_thread_bundle.cpp
+++ b/vespalib/src/vespa/vespalib/util/simple_thread_bundle.cpp
@@ -60,9 +60,10 @@ Signal::Signal() noexcept
 {}
 Signal::~Signal() = default;
 
-SimpleThreadBundle::Pool::Pool(size_t bundleSize)
+SimpleThreadBundle::Pool::Pool(size_t bundleSize, init_fun_t init_fun)
     : _lock(),
       _bundleSize(bundleSize),
+      _init_fun(init_fun),
       _bundles()
 {
 }
@@ -86,7 +87,7 @@ SimpleThreadBundle::Pool::obtain()
             return ret;
         }
     }
-    return std::make_unique<SimpleThreadBundle>(_bundleSize);
+    return std::make_unique<SimpleThreadBundle>(_bundleSize, _init_fun);
 }
 
 void
@@ -99,7 +100,7 @@ SimpleThreadBundle::Pool::release(SimpleThreadBundle::UP bundle)
 
 //-----------------------------------------------------------------------------
 
-SimpleThreadBundle::SimpleThreadBundle(size_t size_in, Strategy strategy)
+SimpleThreadBundle::SimpleThreadBundle(size_t size_in, Runnable::init_fun_t init_fun, Strategy strategy)
     : _work(),
       _signals(),
       _workers(),
@@ -134,7 +135,7 @@ SimpleThreadBundle::SimpleThreadBundle(size_t size_in, Strategy strategy)
             _hook = std::move(hook);
         } else {
             size_t signal_idx = (strategy == USE_BROADCAST) ? 0 : (i - 1);
-            _workers.push_back(std::make_unique<Worker>(_signals[signal_idx], std::move(hook)));
+            _workers.push_back(std::make_unique<Worker>(_signals[signal_idx], init_fun, std::move(hook)));
         }
     }
 }
@@ -175,19 +176,19 @@ SimpleThreadBundle::run(const std::vector<Runnable*> &targets)
     latch.await();
 }
 
-SimpleThreadBundle::Worker::Worker(Signal &s, Runnable::UP h)
-    : thread(*this, simple_thread_bundle_executor),
-      signal(s),
-      hook(std::move(h))
+SimpleThreadBundle::Worker::Worker(Signal &s, Runnable::init_fun_t init_fun, Runnable::UP h)
+  : thread(*this, std::move(init_fun)),
+    signal(s),
+    hook(std::move(h))
 {
     thread.start();
 }
+
 void
 SimpleThreadBundle::Worker::run() {
     for (size_t gen = 0; signal.wait(gen) > 0; ) {
-    hook->run();
-}
-
+        hook->run();
+    }
 }
 
 } // namespace vespalib
diff --git a/vespalib/src/vespa/vespalib/util/simple_thread_bundle.h b/vespalib/src/vespa/vespalib/util/simple_thread_bundle.h
index d9a29ee7bef..b7434d09ac3 100644
--- a/vespalib/src/vespa/vespalib/util/simple_thread_bundle.h
+++ b/vespalib/src/vespa/vespalib/util/simple_thread_bundle.h
@@ -86,8 +86,9 @@ struct Signal {
 class SimpleThreadBundle : public ThreadBundle
 {
 public:
-    typedef fixed_thread_bundle::Work Work;
-    typedef fixed_thread_bundle::Signal Signal;
+    using Work = fixed_thread_bundle::Work;
+    using Signal = fixed_thread_bundle::Signal;
+    using init_fun_t = Runnable::init_fun_t;
 
     typedef std::unique_ptr<SimpleThreadBundle> UP;
     enum Strategy { USE_SIGNAL_LIST, USE_SIGNAL_TREE, USE_BROADCAST };
@@ -97,10 +98,12 @@ public:
     private:
         std::mutex _lock;
         size_t     _bundleSize;
+        init_fun_t _init_fun;
         std::vector<SimpleThreadBundle*> _bundles;
 
     public:
-        Pool(size_t bundleSize);
+        Pool(size_t bundleSize, init_fun_t init_fun);
+        Pool(size_t bundleSize) : Pool(bundleSize, Runnable::default_init_function) {}
         ~Pool();
         SimpleThreadBundle::UP obtain();
         void release(SimpleThreadBundle::UP bundle);
@@ -112,7 +115,7 @@ private:
         Thread thread;
         Signal &signal;
         Runnable::UP hook;
-        Worker(Signal &s, Runnable::UP h);
+        Worker(Signal &s, init_fun_t init_fun, Runnable::UP h);
         void run() override;
     };
 
@@ -122,7 +125,9 @@ private:
     Runnable::UP            _hook;
 
 public:
-    SimpleThreadBundle(size_t size, Strategy strategy = USE_SIGNAL_LIST);
+    SimpleThreadBundle(size_t size, init_fun_t init_fun, Strategy strategy = USE_SIGNAL_LIST);
+    SimpleThreadBundle(size_t size, Strategy strategy = USE_SIGNAL_LIST)
+      : SimpleThreadBundle(size, Runnable::default_init_function, strategy) {}
     ~SimpleThreadBundle();
     size_t size() const override;
     void run(const std::vector<Runnable*> &targets) override;