Add explorer for the ExecutorThreadingService used in a document database.

7 files changed, 168 insertions, 11 deletions

diff --git a/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt b/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt
index 57445775df3..81dd8a64a6c 100644
--- a/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt
+++ b/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt
@@ -36,6 +36,7 @@ vespa_add_library(searchcore_server STATIC
     documentsubdbcollection.cpp
     emptysearchview.cpp
     executor_thread_service.cpp
+    executor_threading_service_explorer.cpp
     executorthreadingservice.cpp
     fast_access_document_retriever.cpp
     fast_access_doc_subdb.cpp
diff --git a/searchcore/src/vespa/searchcore/proton/server/document_db_explorer.cpp b/searchcore/src/vespa/searchcore/proton/server/document_db_explorer.cpp
index 0c3772cab5b..a5c3e8d6078 100644
--- a/searchcore/src/vespa/searchcore/proton/server/document_db_explorer.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/document_db_explorer.cpp
@@ -1,12 +1,12 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "document_db_explorer.h"
-
 #include "document_meta_store_read_guards.h"
 #include "document_subdb_collection_explorer.h"
+#include "executor_threading_service_explorer.h"
 #include "maintenance_controller_explorer.h"
-#include <vespa/searchcore/proton/common/state_reporter_utils.h>
 #include <vespa/searchcore/proton/bucketdb/bucket_db_explorer.h>
+#include <vespa/searchcore/proton/common/state_reporter_utils.h>
 #include <vespa/searchcore/proton/matching/session_manager_explorer.h>
 #include <vespa/vespalib/data/slime/slime.h>
 
@@ -42,6 +42,7 @@ DocumentDBExplorer::get_state(const Inserter &inserter, bool full) const
 }
 
 const vespalib::string SUB_DB = "subdb";
+const vespalib::string THREADING_SERVICE = "threadingservice";
 const vespalib::string BUCKET_DB = "bucketdb";
 const vespalib::string MAINTENANCE_CONTROLLER = "maintenancecontroller";
 const vespalib::string SESSION = "session";
@@ -49,25 +50,24 @@ const vespalib::string SESSION = "session";
 std::vector<vespalib::string>
 DocumentDBExplorer::get_children_names() const
 {
-    return {SUB_DB, BUCKET_DB, MAINTENANCE_CONTROLLER, SESSION};
+    return {SUB_DB, THREADING_SERVICE, BUCKET_DB, MAINTENANCE_CONTROLLER, SESSION};
 }
 
 std::unique_ptr<StateExplorer>
 DocumentDBExplorer::get_child(vespalib::stringref name) const
 {
     if (name == SUB_DB) {
-        return std::unique_ptr<StateExplorer>
-            (new DocumentSubDBCollectionExplorer(_docDb->getDocumentSubDBs()));
+        return std::make_unique<DocumentSubDBCollectionExplorer>(_docDb->getDocumentSubDBs());
+    } else if (name == THREADING_SERVICE) {
+        return std::make_unique<ExecutorThreadingServiceExplorer>(_docDb->getWriteService());
     } else if (name == BUCKET_DB) {
         // TODO(geirst): const_cast can be avoided if we add const guard to BucketDBOwner.
-        return std::unique_ptr<StateExplorer>(new BucketDBExplorer(
-            (const_cast<DocumentSubDBCollection &>(_docDb->getDocumentSubDBs())).getBucketDB().takeGuard()));
+        return std::make_unique<BucketDBExplorer>(
+            (const_cast<DocumentSubDBCollection &>(_docDb->getDocumentSubDBs())).getBucketDB().takeGuard());
     } else if (name == MAINTENANCE_CONTROLLER) {
-        return std::unique_ptr<StateExplorer>
-            (new MaintenanceControllerExplorer(_docDb->getMaintenanceController().getJobList()));
+        return std::make_unique<MaintenanceControllerExplorer>(_docDb->getMaintenanceController().getJobList());
     } else if (name == SESSION) {
-        return std::unique_ptr<StateExplorer>
-            (new matching::SessionManagerExplorer(_docDb->session_manager()));
+        return std::make_unique<matching::SessionManagerExplorer>(_docDb->session_manager());
     }
     return std::unique_ptr<StateExplorer>(nullptr);
 }
diff --git a/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.cpp b/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.cpp
index 24f5fa9a5e6..49dc038096c 100644
--- a/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.cpp
@@ -81,6 +81,10 @@ void ExecutorThreadService::setTaskLimit(uint32_t taskLimit) {
     _executor.setTaskLimit(taskLimit);
 }
 
+uint32_t ExecutorThreadService::getTaskLimit() const {
+    return _executor.getTaskLimit();
+}
+
 void
 ExecutorThreadService::wakeup() {
     _executor.wakeup();
diff --git a/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.h b/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.h
index 4f27a8f86c2..325c36b7270 100644
--- a/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.h
+++ b/searchcore/src/vespa/searchcore/proton/server/executor_thread_service.h
@@ -39,6 +39,7 @@ public:
     size_t getNumThreads() const override { return _executor.getNumThreads(); }
 
     void setTaskLimit(uint32_t taskLimit) override;
+    uint32_t getTaskLimit() const override;
     void wakeup() override;
 };
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/executor_threading_service_explorer.cpp b/searchcore/src/vespa/searchcore/proton/server/executor_threading_service_explorer.cpp
new file mode 100644
index 00000000000..0ecdca54e27
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/server/executor_threading_service_explorer.cpp
@@ -0,0 +1,122 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "executor_threading_service_explorer.h"
+#include "executorthreadingservice.h"
+#include <vespa/vespalib/data/slime/cursor.h>
+#include <vespa/vespalib/util/adaptive_sequenced_executor.h>
+#include <vespa/vespalib/util/blockingthreadstackexecutor.h>
+#include <vespa/vespalib/util/sequencedtaskexecutor.h>
+#include <vespa/vespalib/util/singleexecutor.h>
+#include <vespa/vespalib/util/threadexecutor.h>
+#include <vespa/vespalib/util/threadstackexecutor.h>
+
+using vespalib::AdaptiveSequencedExecutor;
+using vespalib::BlockingThreadStackExecutor;
+using vespalib::ISequencedTaskExecutor;
+using vespalib::SequencedTaskExecutor;
+using vespalib::SingleExecutor;
+using vespalib::SyncableThreadExecutor;
+using vespalib::ThreadStackExecutor;
+using vespalib::slime::Cursor;
+
+namespace proton {
+
+namespace {
+
+void
+set_type(Cursor& object, const vespalib::string& type)
+{
+    object.setString("type", type);
+}
+
+void
+convert_syncable_executor_to_slime(const SyncableThreadExecutor& executor, const vespalib::string& type, Cursor& object)
+{
+    set_type(object, type);
+    object.setLong("num_threads", executor.getNumThreads());
+    object.setLong("task_limit", executor.getTaskLimit());
+}
+
+void
+convert_single_executor_to_slime(const SingleExecutor& executor, Cursor& object)
+{
+    convert_syncable_executor_to_slime(executor, "SingleExecutor", object);
+    object.setLong("watermark", executor.get_watermark());
+    object.setDouble("reaction_time_sec", vespalib::to_s(executor.get_reaction_time()));
+}
+
+void
+convert_executor_to_slime(const SyncableThreadExecutor* executor, Cursor& object)
+{
+    if (executor == nullptr) {
+        return;
+    }
+    if (const auto* single = dynamic_cast<const SingleExecutor*>(executor)) {
+        convert_single_executor_to_slime(*single, object);
+    } else if (const auto* blocking = dynamic_cast<const BlockingThreadStackExecutor*>(executor)) {
+        convert_syncable_executor_to_slime(*blocking, "BlockingThreadStackExecutor", object);
+    } else if (const auto* thread = dynamic_cast<const ThreadStackExecutor*>(executor)) {
+        convert_syncable_executor_to_slime(*thread, "ThreadStackExecutor", object);
+    } else {
+        convert_syncable_executor_to_slime(*executor, "SyncableThreadExecutor", object);
+    }
+}
+
+void
+convert_sequenced_executor_to_slime(const SequencedTaskExecutor& executor, Cursor& object)
+{
+    set_type(object, "SequencedTaskExecutor");
+    object.setLong("num_executors", executor.getNumExecutors());
+    convert_executor_to_slime(executor.first_executor(), object.setObject("executor"));
+}
+
+void
+convert_adaptive_executor_to_slime(const AdaptiveSequencedExecutor& executor, Cursor& object)
+{
+    set_type(object, "AdaptiveSequencedExecutor");
+    object.setLong("num_strands", executor.getNumExecutors());
+    auto cfg = executor.get_config();
+    object.setLong("num_threads", cfg.num_threads);
+    object.setLong("max_waiting", cfg.max_waiting);
+    object.setLong("max_pending", cfg.max_pending);
+    object.setLong("wakeup_limit", cfg.wakeup_limit);
+}
+
+void
+convert_executor_to_slime(const ISequencedTaskExecutor* executor, Cursor& object)
+{
+    if (const auto* seq = dynamic_cast<const SequencedTaskExecutor*>(executor)) {
+        convert_sequenced_executor_to_slime(*seq, object);
+    } else if (const auto* ada = dynamic_cast<const AdaptiveSequencedExecutor*>(executor)) {
+        convert_adaptive_executor_to_slime(*ada, object);
+    } else {
+        set_type(object, "ISequencedTaskExecutor");
+        object.setLong("num_executors", executor->getNumExecutors());
+    }
+}
+
+}
+
+ExecutorThreadingServiceExplorer::ExecutorThreadingServiceExplorer(ExecutorThreadingService& service)
+    : _service(service)
+{
+}
+
+ExecutorThreadingServiceExplorer::~ExecutorThreadingServiceExplorer() = default;
+
+void
+ExecutorThreadingServiceExplorer::get_state(const vespalib::slime::Inserter& inserter, bool full) const
+{
+    auto& object = inserter.insertObject();
+    if (full) {
+        convert_executor_to_slime(&_service.getMasterExecutor(), object.setObject("master"));
+        convert_executor_to_slime(&_service.getIndexExecutor(), object.setObject("index"));
+        convert_executor_to_slime(&_service.getSummaryExecutor(), object.setObject("summary"));
+        convert_executor_to_slime(&_service.indexFieldInverter(), object.setObject("index_field_inverter"));
+        convert_executor_to_slime(&_service.indexFieldWriter(), object.setObject("index_field_writer"));
+        convert_executor_to_slime(&_service.attributeFieldWriter(), object.setObject("attribute_field_writer"));
+    }
+}
+
+}
+
diff --git a/searchcore/src/vespa/searchcore/proton/server/executor_threading_service_explorer.h b/searchcore/src/vespa/searchcore/proton/server/executor_threading_service_explorer.h
new file mode 100644
index 00000000000..14d25add8d6
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/server/executor_threading_service_explorer.h
@@ -0,0 +1,25 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/vespalib/net/state_explorer.h>
+
+namespace proton {
+
+class ExecutorThreadingService;
+
+/**
+ * Class used to explore the state of the ExecutorThreadingService used in a document database.
+ */
+class ExecutorThreadingServiceExplorer : public vespalib::StateExplorer {
+private:
+    ExecutorThreadingService& _service;
+
+public:
+    ExecutorThreadingServiceExplorer(ExecutorThreadingService& service);
+    ~ExecutorThreadingServiceExplorer();
+
+    void get_state(const vespalib::slime::Inserter& inserter, bool full) const override;
+};
+
+}
diff --git a/searchcore/src/vespa/searchcore/proton/test/thread_service_observer.h b/searchcore/src/vespa/searchcore/proton/test/thread_service_observer.h
index 350673b9c90..0b1b9730363 100644
--- a/searchcore/src/vespa/searchcore/proton/test/thread_service_observer.h
+++ b/searchcore/src/vespa/searchcore/proton/test/thread_service_observer.h
@@ -48,6 +48,10 @@ public:
         _service.setTaskLimit(taskLimit);
     }
 
+    uint32_t getTaskLimit() const override {
+        return _service.getTaskLimit();
+    }
+
     void wakeup() override {
         _service.wakeup();
     }