Merge pull request #16179 from vespa-engine/balder/make-persistence-engine-bucket-executor-proxy

Make the PersistenceEngine a proxy and gatekeeper for the BucketExecu…

2 files changed, 24 insertions, 3 deletions

diff --git a/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.cpp b/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.cpp
index 8bec6f9dd68..b188b97404d 100644
--- a/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.cpp
+++ b/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.cpp
@@ -12,7 +12,6 @@
 #include <vespa/document/base/exceptions.h>
 #include <thread>
 
-
 #include <vespa/log/log.h>
 LOG_SETUP(".proton.persistenceengine.persistenceengine");
 
@@ -20,6 +19,7 @@ using document::Document;
 using document::DocumentId;
 using storage::spi::BucketChecksum;
 using storage::spi::BucketExecutor;
+using storage::spi::BucketTask;
 using storage::spi::BucketIdListResult;
 using storage::spi::BucketInfo;
 using storage::spi::BucketInfoResult;
@@ -763,4 +763,20 @@ PersistenceEngine::register_executor(std::shared_ptr<BucketExecutor> executor)
     return std::make_unique<SyncExecutorOnDestruction>(executor);
 }
 
+std::unique_ptr<BucketTask>
+PersistenceEngine::execute(const storage::spi::Bucket &bucket, std::unique_ptr<BucketTask> task) {
+    auto bucketExecutor = get_bucket_executor();
+    if (bucketExecutor) {
+        bucketExecutor->execute(bucket, std::move(task));
+    }
+    return task;
+}
+
+void PersistenceEngine::sync() {
+    auto bucketExecutor = get_bucket_executor();
+    if (bucketExecutor) {
+        bucketExecutor->sync();
+    }
+}
+
 } // storage
diff --git a/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.h b/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.h
index b5a99525575..77dfc74d765 100644
--- a/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.h
+++ b/searchcore/src/vespa/searchcore/proton/persistenceengine/persistenceengine.h
@@ -7,6 +7,7 @@
 #include "ipersistencehandler.h"
 #include "resource_usage_tracker.h"
 #include <vespa/persistence/spi/abstractpersistenceprovider.h>
+#include <vespa/persistence/spi/bucketexecutor.h>
 #include <mutex>
 #include <shared_mutex>
 
@@ -15,7 +16,8 @@ namespace proton {
 class IPersistenceEngineOwner;
 class IDiskMemUsageNotifier;
 
-class PersistenceEngine : public storage::spi::AbstractPersistenceProvider {
+class PersistenceEngine : public storage::spi::AbstractPersistenceProvider,
+                          public storage::spi::BucketExecutor {
 private:
     using PersistenceHandlerSequence = PersistenceHandlerMap::PersistenceHandlerSequence;
     using HandlerSnapshot = PersistenceHandlerMap::HandlerSnapshot;
@@ -40,6 +42,7 @@ private:
     using UpdateResult = storage::spi::UpdateResult;
     using OperationComplete = storage::spi::OperationComplete;
     using BucketExecutor = storage::spi::BucketExecutor;
+    using BucketTask = storage::spi::BucketTask;
 
     struct IteratorEntry {
         PersistenceHandlerSequence  handler_sequence;
@@ -86,6 +89,7 @@ private:
 
     void saveClusterState(BucketSpace bucketSpace, const ClusterState &calc);
     ClusterState::SP savedClusterState(BucketSpace bucketSpace) const;
+    std::shared_ptr<BucketExecutor> get_bucket_executor() noexcept { return _bucket_executor.lock(); }
 
 public:
     typedef std::unique_ptr<PersistenceEngine> UP;
@@ -125,7 +129,8 @@ public:
     void populateInitialBucketDB(const WriteGuard & guard, BucketSpace bucketSpace, IPersistenceHandler &targetHandler);
     WriteGuard getWLock() const;
     ResourceUsageTracker &get_resource_usage_tracker() noexcept { return *_resource_usage_tracker; }
-    std::shared_ptr<BucketExecutor> get_bucket_executor() noexcept { return _bucket_executor.lock(); }
+    std::unique_ptr<BucketTask> execute(const Bucket &bucket, std::unique_ptr<BucketTask> task) override;
+    void sync() override;
 };
 
 }