Merge pull request #19703 from vespa-engine/toregge/forward-new-config-to-merge-handler

Forward new config to merge handler.

5 files changed, 22 insertions, 3 deletions

diff --git a/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp b/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
index 2ffb827accf..c32efb6aa66 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
@@ -209,6 +209,11 @@ FileStorManager::configure(std::unique_ptr<vespa::config::content::StorFilestorC
                                                                    *_filestorHandler, i % numStripes, _component));
         }
         _bucketExecutorRegistration = _provider->register_executor(std::make_shared<BucketExecutorWrapper>(*this));
+    } else {
+        std::lock_guard guard(_lock);        
+        for (auto& handler : _persistenceHandlers) {
+            handler->configure(*config);
+        }
     }
 }
 
diff --git a/storage/src/vespa/storage/persistence/mergehandler.cpp b/storage/src/vespa/storage/persistence/mergehandler.cpp
index 6d58966d415..77e7762ec9a 100644
--- a/storage/src/vespa/storage/persistence/mergehandler.cpp
+++ b/storage/src/vespa/storage/persistence/mergehandler.cpp
@@ -1251,7 +1251,7 @@ MergeHandler::handleApplyBucketDiff(api::ApplyBucketDiffCommand& cmd, MessageTra
        framework::MilliSecTimer startTime(_clock);
        async_results = std::make_shared<ApplyBucketDiffState>(*this, bucket, RetainGuard(*_monitored_ref_count));
        applyDiffLocally(bucket, cmd.getDiff(), index, tracker->context(), async_results);
-        if (!_async_apply_bucket_diff) {
+       if (!_async_apply_bucket_diff.load(std::memory_order_relaxed)) {
             check_apply_diff_sync(std::move(async_results));
         }
         _env._metrics.merge_handler_metrics.mergeDataWriteLatency.addValue(
@@ -1359,7 +1359,7 @@ MergeHandler::handleApplyBucketDiffReply(api::ApplyBucketDiffReply& reply, Messa
                 framework::MilliSecTimer startTime(_clock);
                 async_results = std::make_shared<ApplyBucketDiffState>(*this, bucket, RetainGuard(*_monitored_ref_count));
                 applyDiffLocally(bucket, diff, index, s->context, async_results);
-                if (!_async_apply_bucket_diff) {
+                if (!_async_apply_bucket_diff.load(std::memory_order_relaxed)) {
                     check_apply_diff_sync(std::move(async_results));
                 }
                 _env._metrics.merge_handler_metrics.mergeDataWriteLatency.addValue(startTime.getElapsedTimeAsDouble());
@@ -1446,4 +1446,10 @@ MergeHandler::drain_async_writes()
     }
 }
 
+void
+MergeHandler::configure(bool async_apply_bucket_diff) noexcept
+{
+    _async_apply_bucket_diff.store(async_apply_bucket_diff, std::memory_order_release);
+}
+
 } // storage
diff --git a/storage/src/vespa/storage/persistence/mergehandler.h b/storage/src/vespa/storage/persistence/mergehandler.h
index cc06f769812..17cfb847d2c 100644
--- a/storage/src/vespa/storage/persistence/mergehandler.h
+++ b/storage/src/vespa/storage/persistence/mergehandler.h
@@ -74,6 +74,7 @@ public:
     MessageTrackerUP handleApplyBucketDiff(api::ApplyBucketDiffCommand&, MessageTrackerUP) const;
     void handleApplyBucketDiffReply(api::ApplyBucketDiffReply&, MessageSender&, MessageTrackerUP) const;
     void drain_async_writes();
+    void configure(bool async_apply_bucket_diff) noexcept;
 
 private:
     const framework::Clock   &_clock;
@@ -83,7 +84,7 @@ private:
     std::unique_ptr<vespalib::MonitoredRefCount> _monitored_ref_count;
     const uint32_t            _maxChunkSize;
     const uint32_t            _commonMergeChainOptimalizationMinimumSize;
-    const bool                _async_apply_bucket_diff;
+    std::atomic<bool>         _async_apply_bucket_diff;
 
     /** Returns a reply if merge is complete */
     api::StorageReply::SP processBucketMerge(const spi::Bucket& bucket,
diff --git a/storage/src/vespa/storage/persistence/persistencehandler.cpp b/storage/src/vespa/storage/persistence/persistencehandler.cpp
index 1ef883fc810..aa1a9c136fd 100644
--- a/storage/src/vespa/storage/persistence/persistencehandler.cpp
+++ b/storage/src/vespa/storage/persistence/persistencehandler.cpp
@@ -150,4 +150,10 @@ PersistenceHandler::processLockedMessage(FileStorHandler::LockedMessage lock) co
     }
 }
 
+void
+PersistenceHandler::configure(vespa::config::content::StorFilestorConfig& config) noexcept
+{
+    _mergeHandler.configure(config.asyncApplyBucketDiff);
+}
+
 }
diff --git a/storage/src/vespa/storage/persistence/persistencehandler.h b/storage/src/vespa/storage/persistence/persistencehandler.h
index a92c2dc78ca..c60fb05e56e 100644
--- a/storage/src/vespa/storage/persistence/persistencehandler.h
+++ b/storage/src/vespa/storage/persistence/persistencehandler.h
@@ -35,6 +35,7 @@ public:
     const AsyncHandler & asyncHandler() const { return _asyncHandler; }
     const SplitJoinHandler & splitjoinHandler() const { return _splitJoinHandler; }
     const SimpleMessageHandler & simpleMessageHandler() const { return _simpleHandler; }
+    void configure(vespa::config::content::StorFilestorConfig& config) noexcept;
 private:
     // Message handling functions
     MessageTracker::UP handleCommandSplitByType(api::StorageCommand&, MessageTracker::UP tracker) const;