Merge pull request #20243 from vespa-engine/balder/reduce-use-of-syncable

Remove the need for Syncable

8 files changed, 41 insertions, 27 deletions

diff --git a/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp b/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp
index e3e6ac6321e..b80cd08ae8e 100644
--- a/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp
@@ -234,8 +234,7 @@ MySearchableContext::MySearchableContext(IThreadingService &writeService,
                                          IBucketDBHandlerInitializer & bucketDBHandlerInitializer)
     : _fastUpdCtx(writeService, bucketDB, bucketDBHandlerInitializer),
       _queryLimiter(), _clock(),
-      _ctx(_fastUpdCtx._ctx, _queryLimiter,
-           _clock, dynamic_cast<vespalib::SyncableThreadExecutor &>(writeService.shared()))
+      _ctx(_fastUpdCtx._ctx, _queryLimiter, _clock, writeService.shared())
 {}
 MySearchableContext::~MySearchableContext() = default;
 
diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h b/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h
index 4eaa722e0ba..632f4482654 100644
--- a/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h
+++ b/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h
@@ -119,7 +119,7 @@ public:
     /**
      * Returns the underlying executor. Only used for state explorers.
      */
-    const vespalib::SyncableThreadExecutor& get_executor() const { return _executor; }
+    const vespalib::ThreadExecutor& get_executor() const { return _executor; }
 
     /**
      * Starts the scheduling thread of this manager.
diff --git a/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.h b/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.h
index 7cc0c97048b..3d3be775a4a 100644
--- a/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.h
+++ b/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.h
@@ -69,7 +69,7 @@ public:
     /**
      * Returns the underlying executor. Only used for state explorers.
      */
-    const vespalib::SyncableThreadExecutor& get_executor() const { return _executor; }
+    const vespalib::ThreadExecutor& get_executor() const { return _executor; }
 
     /**
      * Closes the request handler interface. This will prevent any more data
diff --git a/searchcore/src/vespa/searchcore/proton/server/i_proton_configurer_owner.h b/searchcore/src/vespa/searchcore/proton/server/i_proton_configurer_owner.h
index 0d6bb07b173..704b54dc566 100644
--- a/searchcore/src/vespa/searchcore/proton/server/i_proton_configurer_owner.h
+++ b/searchcore/src/vespa/searchcore/proton/server/i_proton_configurer_owner.h
@@ -6,8 +6,6 @@
 #include <vespa/vespalib/stllike/string.h>
 #include <memory>
 
-namespace vespalib { class ThreadStackExecutorBase; }
-
 namespace proton {
 
 class DocumentDBConfigOwner;
@@ -19,7 +17,7 @@ class DocumentDBConfigOwner;
 class IProtonConfigurerOwner
 {
 public:
-    using InitializeThreads = std::shared_ptr<vespalib::SyncableThreadExecutor>;
+    using InitializeThreads = std::shared_ptr<vespalib::ThreadExecutor>;
     virtual ~IProtonConfigurerOwner() { }
     virtual std::shared_ptr<DocumentDBConfigOwner> addDocumentDB(const DocTypeName &docTypeName,
                                                                  document::BucketSpace bucketSpace,
diff --git a/searchcore/src/vespa/searchcore/proton/server/proton.h b/searchcore/src/vespa/searchcore/proton/server/proton.h
index c57d3e26d8b..91635dc7497 100644
--- a/searchcore/src/vespa/searchcore/proton/server/proton.h
+++ b/searchcore/src/vespa/searchcore/proton/server/proton.h
@@ -59,8 +59,7 @@ private:
     using MonitorReply = search::engine::MonitorReply;
     using MonitorClient = search::engine::MonitorClient;
     using DocumentDBMap = std::map<DocTypeName, DocumentDB::SP>;
-    using ProtonConfigSP = BootstrapConfig::ProtonConfigSP;
-    using InitializeThreads = std::shared_ptr<vespalib::SyncableThreadExecutor>;
+    using InitializeThreads = std::shared_ptr<vespalib::ThreadExecutor>;
     using BucketSpace = document::BucketSpace;
 
     class ProtonFileHeaderContext : public search::common::FileHeaderContext
diff --git a/searchcore/src/vespa/searchcore/proton/server/proton_configurer.cpp b/searchcore/src/vespa/searchcore/proton/server/proton_configurer.cpp
index 7c998ceca7c..2c891927fa3 100644
--- a/searchcore/src/vespa/searchcore/proton/server/proton_configurer.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/proton_configurer.cpp
@@ -12,6 +12,7 @@
 #include <vespa/document/bucket/fixed_bucket_spaces.h>
 #include <vespa/config-bucketspaces.h>
 #include <vespa/vespalib/util/exceptions.h>
+#include <vespa/vespalib/util/retain_guard.h>
 #include <vespa/vespalib/stllike/asciistream.h>
 #include <future>
 
@@ -42,7 +43,7 @@ getBucketSpace(const BootstrapConfig &bootstrapConfig, const DocTypeName &name)
 }
 
 
-ProtonConfigurer::ProtonConfigurer(vespalib::SyncableThreadExecutor &executor,
+ProtonConfigurer::ProtonConfigurer(vespalib::ThreadExecutor &executor,
                                    IProtonConfigurerOwner &owner,
                                    const std::unique_ptr<IProtonDiskLayout> &diskLayout)
     : IProtonConfigurer(),
@@ -58,9 +59,22 @@ ProtonConfigurer::ProtonConfigurer(vespalib::SyncableThreadExecutor &executor,
 {
 }
 
-ProtonConfigurer::~ProtonConfigurer()
-{
-}
+class ProtonConfigurer::ReconfigureTask : public vespalib::Executor::Task {
+public:
+    ReconfigureTask(ProtonConfigurer & configurer)
+        : _configurer(configurer),
+          _retainGuard(configurer._pendingReconfigureTasks)
+    {}
+
+    void run() override {
+        _configurer.performReconfigure();
+    }
+private:
+    ProtonConfigurer      & _configurer;
+    vespalib::RetainGuard   _retainGuard;
+};
+
+ProtonConfigurer::~ProtonConfigurer() = default;
 
 void
 ProtonConfigurer::setAllowReconfig(bool allowReconfig)
@@ -72,11 +86,12 @@ ProtonConfigurer::setAllowReconfig(bool allowReconfig)
         _allowReconfig = allowReconfig;
         if (allowReconfig) {
             // Ensure that pending config is applied
-            _executor.execute(makeLambdaTask([this]() { performReconfigure(); }));
+            _executor.execute(std::make_unique<ReconfigureTask>(*this));
         }
     }
     if (!allowReconfig) {
-        _executor.sync(); // drain queued performReconfigure tasks
+        // drain queued performReconfigure tasks
+        _pendingReconfigureTasks.waitForZeroRefCount();
     }
 }
 
@@ -102,7 +117,7 @@ ProtonConfigurer::reconfigure(std::shared_ptr<ProtonConfigSnapshot> configSnapsh
     std::lock_guard<std::mutex> guard(_mutex);
     _pendingConfigSnapshot = configSnapshot;
     if (_allowReconfig) {
-        _executor.execute(makeLambdaTask([&]() { performReconfigure(); }));
+        _executor.execute(std::make_unique<ReconfigureTask>(*this));
     }
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/proton_configurer.h b/searchcore/src/vespa/searchcore/proton/server/proton_configurer.h
index 3ebfbc378d7..ddb9c1bed92 100644
--- a/searchcore/src/vespa/searchcore/proton/server/proton_configurer.h
+++ b/searchcore/src/vespa/searchcore/proton/server/proton_configurer.h
@@ -7,6 +7,7 @@
 #include <vespa/document/bucket/bucketspace.h>
 #include <vespa/searchcore/proton/common/doctypename.h>
 #include <vespa/vespalib/net/simple_component_config_producer.h>
+#include <vespa/vespalib/util/monitored_refcount.h>
 #include <map>
 #include <mutex>
 
@@ -25,17 +26,19 @@ class IProtonDiskLayout;
 class ProtonConfigurer : public IProtonConfigurer
 {
     using DocumentDBs = std::map<DocTypeName, std::pair<std::weak_ptr<IDocumentDBConfigOwner>, std::weak_ptr<DocumentDBDirectoryHolder>>>;
-    using InitializeThreads = std::shared_ptr<vespalib::SyncableThreadExecutor>;
+    using InitializeThreads = std::shared_ptr<vespalib::ThreadExecutor>;
+    class ReconfigureTask;
 
-    SyncableExecutorThreadService _executor;
-    IProtonConfigurerOwner &_owner;
-    DocumentDBs _documentDBs;
-    std::shared_ptr<ProtonConfigSnapshot> _pendingConfigSnapshot;
-    std::shared_ptr<ProtonConfigSnapshot> _activeConfigSnapshot;
-    mutable std::mutex _mutex;
-    bool _allowReconfig;
-    vespalib::SimpleComponentConfigProducer _componentConfig;
+    ExecutorThreadService                     _executor;
+    IProtonConfigurerOwner                   &_owner;
+    DocumentDBs                               _documentDBs;
+    std::shared_ptr<ProtonConfigSnapshot>     _pendingConfigSnapshot;
+    std::shared_ptr<ProtonConfigSnapshot>     _activeConfigSnapshot;
+    mutable std::mutex                        _mutex;
+    bool                                      _allowReconfig;
+    vespalib::SimpleComponentConfigProducer   _componentConfig;
     const std::unique_ptr<IProtonDiskLayout> &_diskLayout;
+    vespalib::MonitoredRefCount               _pendingReconfigureTasks;
 
     void performReconfigure();
     bool skipConfig(const ProtonConfigSnapshot *configSnapshot, bool initialConfig);
@@ -48,7 +51,7 @@ class ProtonConfigurer : public IProtonConfigurer
     void pruneInitialDocumentDBDirs(const ProtonConfigSnapshot &configSnapshot);
 
 public:
-    ProtonConfigurer(vespalib::SyncableThreadExecutor &executor,
+    ProtonConfigurer(vespalib::ThreadExecutor &executor,
                      IProtonConfigurerOwner &owner,
                      const std::unique_ptr<IProtonDiskLayout> &diskLayout);
 
diff --git a/searchcore/src/vespa/searchcore/proton/summaryengine/summaryengine.h b/searchcore/src/vespa/searchcore/proton/summaryengine/summaryengine.h
index 7f6d9328491..c49649de1e3 100644
--- a/searchcore/src/vespa/searchcore/proton/summaryengine/summaryengine.h
+++ b/searchcore/src/vespa/searchcore/proton/summaryengine/summaryengine.h
@@ -72,7 +72,7 @@ public:
     /**
      * Returns the underlying executor. Only used for state explorers.
      */
-    const vespalib::SyncableThreadExecutor& get_executor() const { return _executor; }
+    const vespalib::ThreadExecutor& get_executor() const { return _executor; }
 
     /**
      * Starts the underlying threads. This will throw a vespalib::Exception if