Add and wire live config for selecting `DeleteBucket` behavior

By default the legacy behavior is used.

4 files changed, 45 insertions, 9 deletions

diff --git a/storage/src/tests/persistence/filestorage/filestormanagertest.cpp b/storage/src/tests/persistence/filestorage/filestormanagertest.cpp
index 2434c6d3e0d..96618eb9206 100644
--- a/storage/src/tests/persistence/filestorage/filestormanagertest.cpp
+++ b/storage/src/tests/persistence/filestorage/filestormanagertest.cpp
@@ -45,6 +45,8 @@ using namespace storage::api;
 using storage::spi::test::makeSpiBucket;
 using document::test::makeDocumentBucket;
 using vespalib::IDestructorCallback;
+using vespa::config::content::StorFilestorConfig;
+using vespa::config::content::StorFilestorConfigBuilder;
 using namespace ::testing;
 
 #define ASSERT_SINGLE_REPLY(replytype, reply, link, time) \
@@ -224,7 +226,6 @@ struct TestFileStorComponents {
     explicit TestFileStorComponents(FileStorTestBase& test, bool use_small_config = false)
         : manager(nullptr)
     {
-        using vespa::config::content::StorFilestorConfig;
         auto config_uri = config::ConfigUri((use_small_config ? test.smallConfig : test.config)->getConfigId());
         auto config = config_from<StorFilestorConfig>(config_uri);
         auto fsm = std::make_unique<FileStorManager>(*config, test._node->getPersistenceProvider(),
@@ -275,7 +276,7 @@ struct PersistenceHandlerComponents : public FileStorHandlerComponents {
           bucketOwnershipNotifier(component, messageSender),
           persistenceHandler()
     {
-        vespa::config::content::StorFilestorConfig cfg;
+        StorFilestorConfig cfg;
         persistenceHandler =
                 std::make_unique<PersistenceHandler>(executor, component, cfg,
                                                      test._node->getPersistenceProvider(),
@@ -310,7 +311,7 @@ FileStorTestBase::TearDown()
 }
 
 struct FileStorManagerTest : public FileStorTestBase {
-
+    void do_test_delete_bucket(bool use_throttled_delete);
 };
 
 TEST_F(FileStorManagerTest, header_only_put) {
@@ -743,7 +744,7 @@ TEST_F(FileStorManagerTest, priority) {
 
     ServiceLayerComponent component(_node->getComponentRegister(), "test");
     BucketOwnershipNotifier bucketOwnershipNotifier(component, c.messageSender);
-    vespa::config::content::StorFilestorConfig cfg;
+    StorFilestorConfig cfg;
     PersistenceHandler persistenceHandler(_node->executor(), component, cfg, _node->getPersistenceProvider(),
                                           filestorHandler, bucketOwnershipNotifier, *metrics.threads[0]);
     std::unique_ptr<DiskThread> thread(createThread(persistenceHandler, filestorHandler, component));
@@ -1370,8 +1371,14 @@ TEST_F(FileStorManagerTest, remove_location) {
     }
 }
 
-TEST_F(FileStorManagerTest, delete_bucket) {
+void FileStorManagerTest::do_test_delete_bucket(bool use_throttled_delete) {
     TestFileStorComponents c(*this);
+
+    auto config_uri = config::ConfigUri(config->getConfigId());
+    StorFilestorConfigBuilder my_config(*config_from<StorFilestorConfig>(config_uri));
+    my_config.usePerDocumentThrottledDeleteBucket = use_throttled_delete;
+    c.manager->on_configure(my_config);
+
     auto& top = c.top;
     // Creating a document to test with
     document::DocumentId docId("id:crawler:testdoctype1:n=4000:http://www.ntnu.no/");
@@ -1416,6 +1423,15 @@ TEST_F(FileStorManagerTest, delete_bucket) {
     }
 }
 
+// TODO remove once throttled behavior is the default
+TEST_F(FileStorManagerTest, delete_bucket_legacy) {
+    do_test_delete_bucket(false);
+}
+
+TEST_F(FileStorManagerTest, delete_bucket_throttled) {
+    do_test_delete_bucket(true);
+}
+
 TEST_F(FileStorManagerTest, delete_bucket_rejects_outdated_bucket_info) {
     TestFileStorComponents c(*this);
     auto& top = c.top;
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp b/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
index e6cd8987c0a..61c7da6e286 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
@@ -230,7 +230,7 @@ FileStorManager::on_configure(const StorFilestorConfig& config)
                                                                      numResponseThreads, 10000,
                                                                      true, selectSequencer(_config->responseSequencerType));
         assert(_sequencedExecutor);
-        LOG(spam, "Setting up the disk");
+        LOG(spam, "Setting up %u persistence threads", numThreads);
         for (uint32_t i = 0; i < numThreads; i++) {
             _threads.push_back(std::make_unique<PersistenceThread>(createRegisteredHandler(_component),
                                                                    *_filestorHandler, i % numStripes, _component));
@@ -241,13 +241,14 @@ FileStorManager::on_configure(const StorFilestorConfig& config)
         auto updated_dyn_throttle_params = dynamic_throttle_params_from_config(config, _threads.size());
         _filestorHandler->reconfigure_dynamic_throttler(updated_dyn_throttle_params);
     }
-    // TODO remove once desired dynamic throttling behavior is set in stone
+    // TODO remove once desired throttling behavior is set in stone
     {
         _filestorHandler->use_dynamic_operation_throttling(use_dynamic_throttling);
         _filestorHandler->set_throttle_apply_bucket_diff_ops(!throttle_merge_feed_ops);
         std::lock_guard guard(_lock);
         for (auto& ph : _persistenceHandlers) {
             ph->set_throttle_merge_feed_ops(throttle_merge_feed_ops);
+            ph->set_use_per_document_throttled_delete_bucket(config.usePerDocumentThrottledDeleteBucket);
         }
     }
 }
diff --git a/storage/src/vespa/storage/persistence/persistencehandler.cpp b/storage/src/vespa/storage/persistence/persistencehandler.cpp
index 33171296abf..78ad6eac0e2 100644
--- a/storage/src/vespa/storage/persistence/persistencehandler.cpp
+++ b/storage/src/vespa/storage/persistence/persistencehandler.cpp
@@ -24,7 +24,8 @@ PersistenceHandler::PersistenceHandler(vespalib::ISequencedTaskExecutor & sequen
                     cfg.commonMergeChainOptimalizationMinimumSize),
       _asyncHandler(_env, provider, bucketOwnershipNotifier, sequencedExecutor, component.getBucketIdFactory()),
       _splitJoinHandler(_env, provider, bucketOwnershipNotifier, cfg.enableMultibitSplitOptimalization),
-      _simpleHandler(_env, provider, component.getBucketIdFactory())
+      _simpleHandler(_env, provider, component.getBucketIdFactory()),
+      _use_per_op_throttled_delete_bucket(false)
 {
 }
 
@@ -68,7 +69,11 @@ PersistenceHandler::handleCommandSplitByType(api::StorageCommand& msg, MessageTr
     case api::MessageType::CREATEBUCKET_ID:
         return _asyncHandler.handleCreateBucket(static_cast<api::CreateBucketCommand&>(msg), std::move(tracker));
     case api::MessageType::DELETEBUCKET_ID:
-        return _asyncHandler.handleDeleteBucket(static_cast<api::DeleteBucketCommand&>(msg), std::move(tracker));
+        if (use_per_op_throttled_delete_bucket()) {
+            return _asyncHandler.handle_delete_bucket_throttling(static_cast<api::DeleteBucketCommand&>(msg), std::move(tracker));
+        } else {
+            return _asyncHandler.handleDeleteBucket(static_cast<api::DeleteBucketCommand&>(msg), std::move(tracker));
+        }
     case api::MessageType::JOINBUCKETS_ID:
         return _splitJoinHandler.handleJoinBuckets(static_cast<api::JoinBucketsCommand&>(msg), std::move(tracker));
     case api::MessageType::SPLITBUCKET_ID:
@@ -181,4 +186,14 @@ PersistenceHandler::set_throttle_merge_feed_ops(bool throttle) noexcept
     _mergeHandler.set_throttle_merge_feed_ops(throttle);
 }
 
+bool
+PersistenceHandler::use_per_op_throttled_delete_bucket() const noexcept {
+    return _use_per_op_throttled_delete_bucket.load(std::memory_order_relaxed);
+}
+
+void
+PersistenceHandler::set_use_per_document_throttled_delete_bucket(bool throttle) noexcept {
+    _use_per_op_throttled_delete_bucket.store(throttle, std::memory_order_relaxed);
+}
+
 }
diff --git a/storage/src/vespa/storage/persistence/persistencehandler.h b/storage/src/vespa/storage/persistence/persistencehandler.h
index d7ebec85549..9639b772a28 100644
--- a/storage/src/vespa/storage/persistence/persistencehandler.h
+++ b/storage/src/vespa/storage/persistence/persistencehandler.h
@@ -11,6 +11,7 @@
 #include <vespa/storage/common/storagecomponent.h>
 #include <vespa/vespalib/util/isequencedtaskexecutor.h>
 #include <vespa/config-stor-filestor.h>
+#include <atomic>
 
 namespace storage {
 
@@ -37,12 +38,14 @@ public:
     const SimpleMessageHandler & simpleMessageHandler() const { return _simpleHandler; }
 
     void set_throttle_merge_feed_ops(bool throttle) noexcept;
+    void set_use_per_document_throttled_delete_bucket(bool throttle) noexcept;
 private:
     // Message handling functions
     MessageTracker::UP handleCommandSplitByType(api::StorageCommand&, MessageTracker::UP tracker) const;
     MessageTracker::UP handleReply(api::StorageReply&, MessageTracker::UP) const;
 
     MessageTracker::UP processMessage(api::StorageMessage& msg, MessageTracker::UP tracker) const;
+    [[nodiscard]] bool use_per_op_throttled_delete_bucket() const noexcept;
 
     const framework::Clock  & _clock;
     PersistenceUtil           _env;
@@ -51,6 +54,7 @@ private:
     AsyncHandler              _asyncHandler;
     SplitJoinHandler          _splitJoinHandler;
     SimpleMessageHandler      _simpleHandler;
+    std::atomic<bool>         _use_per_op_throttled_delete_bucket;
 };
 
 } // storage