Improve tracking of remove batch rate used to consider to block lid space compaction.

This is also a preparation for tracking the rate of regular remove operations,
and use this to consider to block lid space compaction.

5 files changed, 150 insertions, 34 deletions

diff --git a/searchcore/src/tests/proton/common/operation_rate_tracker/CMakeLists.txt b/searchcore/src/tests/proton/common/operation_rate_tracker/CMakeLists.txt
new file mode 100644
index 00000000000..f5e6a791124
--- /dev/null
+++ b/searchcore/src/tests/proton/common/operation_rate_tracker/CMakeLists.txt
@@ -0,0 +1,9 @@
+# Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(searchcore_operation_rate_tracker_test_app TEST
+    SOURCES
+    operation_rate_tracker_test.cpp
+    DEPENDS
+    searchcore_pcommon
+    gtest
+)
+vespa_add_test(NAME searchcore_operation_rate_tracker_test_app COMMAND searchcore_operation_rate_tracker_test_app)
diff --git a/searchcore/src/tests/proton/common/operation_rate_tracker/operation_rate_tracker_test.cpp b/searchcore/src/tests/proton/common/operation_rate_tracker/operation_rate_tracker_test.cpp
new file mode 100644
index 00000000000..ce19867e286
--- /dev/null
+++ b/searchcore/src/tests/proton/common/operation_rate_tracker/operation_rate_tracker_test.cpp
@@ -0,0 +1,90 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/searchcore/proton/common/operation_rate_tracker.h>
+#include <vespa/vespalib/gtest/gtest.h>
+#include <vespa/vespalib/util/time.h>
+
+#include <vespa/log/log.h>
+LOG_SETUP("operation_rate_tracker_test");
+
+using namespace proton;
+
+TEST(OperationRateTrackerTest, time_budget_per_op_is_inverse_of_rate_threshold)
+{
+    EXPECT_EQ(vespalib::from_s(0.25), OperationRateTracker(4).get_time_budget_per_op());
+    EXPECT_EQ(vespalib::from_s(2.0), OperationRateTracker(0.5).get_time_budget_per_op());
+}
+
+TEST(OperationRateTrackerTest, time_budget_window_is_minimum_1_sec)
+{
+    EXPECT_EQ(vespalib::from_s(1.0), OperationRateTracker(4).get_time_budget_window());
+    EXPECT_EQ(vespalib::from_s(2.0), OperationRateTracker(0.5).get_time_budget_window());
+}
+
+class Simulator {
+public:
+    vespalib::steady_time now;
+    OperationRateTracker ort;
+    Simulator(double rate_threshold)
+        : now(vespalib::steady_clock::now()),
+          ort(rate_threshold)
+    {
+    }
+    void tick(double real_rate) {
+        now = now + vespalib::from_s(1.0 / real_rate);
+        ort.observe(now);
+    }
+    bool above_threshold(double now_delta = 0) {
+        return ort.above_threshold(now + vespalib::from_s(now_delta));
+    }
+};
+
+TEST(OperationRateTrackerTest, tracks_whether_operation_rate_is_below_or_above_threshold)
+{
+    Simulator sim(2);
+
+    // Simulate an actual rate of 4 ops / sec
+    sim.tick(4); // Threshold time is 1.0s in the past (at time budget window start)
+    EXPECT_FALSE(sim.above_threshold(-1.0));
+    EXPECT_TRUE(sim.above_threshold(-1.01));
+
+    // Catch up with now
+    sim.tick(4);
+    sim.tick(4);
+    sim.tick(4);
+    sim.tick(4); // Threshold time is now.
+    EXPECT_FALSE(sim.above_threshold());
+    EXPECT_TRUE(sim.above_threshold(-0.01));
+
+    // Move into the future
+    sim.tick(4); // Threshold time is 0.25s into the future.
+    EXPECT_TRUE(sim.above_threshold(0.24));
+    EXPECT_FALSE(sim.above_threshold(0.25));
+
+    // Move to time budget window end
+    sim.tick(4);
+    sim.tick(4);
+    sim.tick(4); // Threshold time is 1.0s into the future (at time budget window end)
+    EXPECT_TRUE(sim.above_threshold(0.99));
+    EXPECT_FALSE(sim.above_threshold(1.0));
+
+    sim.tick(4); // Threshold time is still 1.0s into the future (at time budget window end)
+    EXPECT_TRUE(sim.above_threshold(0.99));
+    EXPECT_FALSE(sim.above_threshold(1.0));
+
+    // Reduce actual rate to 1 ops / sec
+    sim.tick(1); // Threshold time is 0.5s into the future.
+    EXPECT_TRUE(sim.above_threshold(0.49));
+    EXPECT_FALSE(sim.above_threshold(0.5));
+
+    sim.tick(1); // Threshold time is now.
+    EXPECT_FALSE(sim.above_threshold());
+    EXPECT_TRUE(sim.above_threshold(-0.01));
+
+    sim.tick(1);
+    sim.tick(1); // Threshold time is back at time budget window start
+    EXPECT_FALSE(sim.above_threshold(-1.0));
+    EXPECT_TRUE(sim.above_threshold(-1.01));
+}
+
+GTEST_MAIN_RUN_ALL_TESTS()
diff --git a/searchcore/src/tests/proton/documentdb/lid_space_compaction/lid_space_compaction_test.cpp b/searchcore/src/tests/proton/documentdb/lid_space_compaction/lid_space_compaction_test.cpp
index 64299c70588..a97a2380f25 100644
--- a/searchcore/src/tests/proton/documentdb/lid_space_compaction/lid_space_compaction_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/lid_space_compaction/lid_space_compaction_test.cpp
@@ -30,7 +30,7 @@ constexpr uint32_t SUBDB_ID = 2;
 constexpr vespalib::duration JOB_DELAY = 1s;
 constexpr uint32_t ALLOWED_LID_BLOAT = 1;
 constexpr double ALLOWED_LID_BLOAT_FACTOR = 0.3;
-constexpr vespalib::duration REMOVE_BATCH_BLOCK_DELAY = 20s;
+constexpr double REMOVE_BATCH_BLOCK_RATE = 1.0 / 20.0;
 constexpr uint32_t MAX_DOCS_TO_SCAN = 100;
 constexpr double RESOURCE_LIMIT_FACTOR = 1.0;
 constexpr uint32_t MAX_OUTSTANDING_MOVE_OPS = 10;
@@ -82,19 +82,30 @@ struct MyHandler : public ILidSpaceCompactionHandler {
     mutable uint32_t _iteratorCnt;
     bool _storeMoveDoneContexts;
     std::vector<IDestructorCallback::SP> _moveDoneContexts;
+    documentmetastore::OperationListener::SP _op_listener;
+    RemoveOperationsRateTracker* _rm_listener;
 
     MyHandler(bool storeMoveDoneContexts = false);
     ~MyHandler();
     void clearMoveDoneContexts() { _moveDoneContexts.clear(); }
-    void set_last_remove_batch(TimePoint last_remove_batch) {
-        for (auto& s : _stats) {
-            s = LidUsageStats(s.getLidLimit(), s.getUsedLids(),
-                              s.getLowestFreeLid(), s.getHighestUsedLid(), last_remove_batch);
-        }
+    void run_remove_batch_ops() {
+        // This ensures to max out the threshold time in the operation rate tracker.
+        _op_listener->notify_remove_batch();
+        _op_listener->notify_remove_batch();
+        _op_listener->notify_remove_batch();
+    }
+    void stop_remove_batch_ops() {
+        _rm_listener->reset_remove_batch_tracker();
     }
     virtual vespalib::string getName() const override {
         return "myhandler";
     }
+    virtual void set_operation_listener(documentmetastore::OperationListener::SP op_listener) override {
+        auto* rm_listener = dynamic_cast<RemoveOperationsRateTracker*>(op_listener.get());
+        assert(rm_listener != nullptr);
+        _op_listener = std::move(op_listener);
+        _rm_listener = rm_listener;
+    }
     virtual uint32_t getSubDbId() const override { return 2; }
     virtual LidUsageStats getLidStatus() const override {
         assert(_handleMoveCnt < _stats.size());
@@ -132,7 +143,9 @@ MyHandler::MyHandler(bool storeMoveDoneContexts)
       _wantedLidLimit(0),
       _iteratorCnt(0),
       _storeMoveDoneContexts(storeMoveDoneContexts),
-      _moveDoneContexts()
+      _moveDoneContexts(),
+      _op_listener(),
+      _rm_listener()
 {}
 
 MyHandler::~MyHandler() {}
@@ -265,7 +278,7 @@ struct JobTestBase : public ::testing::Test {
         _handler = std::make_unique<MyHandler>(maxOutstandingMoveOps != MAX_OUTSTANDING_MOVE_OPS);
         _job = std::make_unique<LidSpaceCompactionJob>(DocumentDBLidSpaceCompactionConfig(interval, allowedLidBloat,
                                                                                           allowedLidBloatFactor,
-                                                                                          REMOVE_BATCH_BLOCK_DELAY,
+                                                                                          REMOVE_BATCH_BLOCK_RATE,
                                                                                           false, maxDocsToScan),
                                                        *_handler, _storer, _frozenHandler, _diskMemUsageNotifier,
                                                        BlockableMaintenanceJobConfig(resourceLimitFactor, maxOutstandingMoveOps),
@@ -274,20 +287,18 @@ struct JobTestBase : public ::testing::Test {
     ~JobTestBase();
     JobTestBase &addStats(uint32_t docIdLimit,
                           const LidVector &usedLids,
-                          const LidPairVector &usedFreePairs,
-                          TimePoint last_remove_batch = TimePoint()) {
-        return addMultiStats(docIdLimit, {usedLids}, usedFreePairs, last_remove_batch);
+                          const LidPairVector &usedFreePairs) {
+        return addMultiStats(docIdLimit, {usedLids}, usedFreePairs);
     }
     JobTestBase &addMultiStats(uint32_t docIdLimit,
                               const std::vector<LidVector> &usedLidsVector,
-                              const LidPairVector &usedFreePairs,
-                              TimePoint last_remove_batch = TimePoint()) {
+                              const LidPairVector &usedFreePairs) {
         uint32_t usedLids = usedLidsVector[0].size();
         for (auto pair : usedFreePairs) {
             uint32_t highestUsedLid = pair.first;
             uint32_t lowestFreeLid = pair.second;
             _handler->_stats.push_back(LidUsageStats
-                    (docIdLimit, usedLids, lowestFreeLid, highestUsedLid, last_remove_batch));
+                    (docIdLimit, usedLids, lowestFreeLid, highestUsedLid));
         }
         _handler->_lids = usedLidsVector;
         return *this;
@@ -297,7 +308,7 @@ struct JobTestBase : public ::testing::Test {
                           uint32_t lowestFreeLid,
                           uint32_t highestUsedLid) {
         _handler->_stats.push_back(LidUsageStats
-                (docIdLimit, numDocs, lowestFreeLid, highestUsedLid, TimePoint()));
+                (docIdLimit, numDocs, lowestFreeLid, highestUsedLid));
         return *this;
     }
     bool run() {
@@ -332,11 +343,10 @@ struct JobTestBase : public ::testing::Test {
     void assertNoWorkDone() {
         assertJobContext(0, 0, 0, 0, 0);
     }
-    JobTestBase &setupOneDocumentToCompact(TimePoint last_remove_batch = TimePoint()) {
+    JobTestBase &setupOneDocumentToCompact() {
         addStats(10, {1,3,4,5,6,9},
                  {{9,2},   // 30% bloat: move 9 -> 2
-                  {6,7}}, // no documents to move
-                 last_remove_batch);
+                  {6,7}}); // no documents to move
         return *this;
     }
     void assertOneDocumentCompacted() {
@@ -622,8 +632,8 @@ TEST_F(JobTest, job_is_re_enabled_when_node_is_no_longer_retired)
 
 TEST_F(JobTest, job_is_disabled_while_remove_batch_is_ongoing)
 {
-    TimePoint last_remove_batch = std::chrono::steady_clock::now();
-    setupOneDocumentToCompact(last_remove_batch);
+    setupOneDocumentToCompact();
+    _handler->run_remove_batch_ops();
     EXPECT_TRUE(run()); // job is disabled
     assertNoWorkDone();
 }
@@ -634,7 +644,7 @@ TEST_F(JobTest, job_becomes_disabled_if_remove_batch_starts)
     EXPECT_FALSE(run()); // job executed as normal (with more work to do)
     assertJobContext(2, 9, 1, 0, 0);
 
-    _handler->set_last_remove_batch(std::chrono::steady_clock::now());
+    _handler->run_remove_batch_ops();
     EXPECT_TRUE(run()); // job is disabled
     assertJobContext(2, 9, 1, 0, 0);
 }
@@ -645,12 +655,11 @@ TEST_F(JobTest, job_is_re_enabled_when_remove_batch_is_no_longer_ongoing)
     EXPECT_FALSE(run()); // job executed as normal (with more work to do)
     assertJobContext(2, 9, 1, 0, 0);
 
-    TimePoint last_remove_batch = std::chrono::steady_clock::now();
-    _handler->set_last_remove_batch(last_remove_batch);
+    _handler->run_remove_batch_ops();
     EXPECT_TRUE(run()); // job is disabled
     assertJobContext(2, 9, 1, 0, 0);
 
-    _handler->set_last_remove_batch(last_remove_batch - REMOVE_BATCH_BLOCK_DELAY);
+    _handler->stop_remove_batch_ops();
     EXPECT_FALSE(run()); // job executed as normal (with more work to do)
     assertJobContext(3, 8, 2, 0, 0);
 }
diff --git a/searchcore/src/tests/proton/documentdb/maintenancecontroller/maintenancecontroller_test.cpp b/searchcore/src/tests/proton/documentdb/maintenancecontroller/maintenancecontroller_test.cpp
index 7e2e258476f..2bd67f96b57 100644
--- a/searchcore/src/tests/proton/documentdb/maintenancecontroller/maintenancecontroller_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/maintenancecontroller/maintenancecontroller_test.cpp
@@ -339,6 +339,7 @@ struct MockLidSpaceCompactionHandler : public ILidSpaceCompactionHandler
 
     MockLidSpaceCompactionHandler(const vespalib::string &name_) : name(name_) {}
     virtual vespalib::string getName() const override { return name; }
+    virtual void set_operation_listener(documentmetastore::OperationListener::SP) override {}
     virtual uint32_t getSubDbId() const override { return 0; }
     virtual search::LidUsageStats getLidStatus() const override { return search::LidUsageStats(); }
     virtual IDocumentScanIterator::UP getIterator() const override { return IDocumentScanIterator::UP(); }
diff --git a/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp b/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp
index f8069dcf494..8935143a7be 100644
--- a/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp
+++ b/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp
@@ -6,6 +6,7 @@
 #include <vespa/searchcore/proton/bucketdb/i_bucket_create_listener.h>
 #include <vespa/searchcore/proton/common/hw_info.h>
 #include <vespa/searchcore/proton/documentmetastore/documentmetastore.h>
+#include <vespa/searchcore/proton/documentmetastore/operation_listener.h>
 #include <vespa/searchcore/proton/flushengine/shrink_lid_space_flush_target.h>
 #include <vespa/searchcore/proton/server/itlssyncer.h>
 #include <vespa/searchlib/attribute/attributefilesavetarget.h>
@@ -1782,7 +1783,7 @@ TEST(DocumentMetaStoreTest, get_lid_usage_stats_works)
 void
 assertLidBloat(uint32_t expBloat, uint32_t lidLimit, uint32_t usedLids)
 {
-    LidUsageStats stats(lidLimit, usedLids, 0, 0, LidUsageStats::TimePoint());
+    LidUsageStats stats(lidLimit, usedLids, 0, 0);
     EXPECT_EQ(expBloat, stats.getLidBloat());
 }
 
@@ -2084,21 +2085,27 @@ TEST(DocumentMetaStoreTest, multiple_lids_can_be_removed_with_removeBatch)
     assertLidGidFound(4, dms);
 }
 
-TEST(DocumentMetaStoreTest, tracks_time_of_last_call_to_remove_batch)
+class MockOperationListener : public documentmetastore::OperationListener {
+public:
+    size_t remove_batch_cnt;
+
+    MockOperationListener()
+        : remove_batch_cnt(0)
+    {
+    }
+    void notify_remove_batch() override { ++remove_batch_cnt; }
+};
+
+TEST(DocumentMetaStoreTest, call_to_remove_batch_is_notified)
 {
     DocumentMetaStore dms(createBucketDB());
+    auto listener = std::make_shared<MockOperationListener>();
+    dms.set_operation_listener(listener);
     dms.constructFreeList();
     addLid(dms, 1);
 
-    LidUsageStats::TimePoint before = std::chrono::steady_clock::now();
-    std::this_thread::sleep_for(std::chrono::milliseconds(1));
     dms.removeBatch({1}, 5);
-    std::this_thread::sleep_for(std::chrono::milliseconds(1));
-    LidUsageStats::TimePoint after = std::chrono::steady_clock::now();
-
-    auto stats = dms.getLidUsageStats();
-    EXPECT_LT(before, stats.get_last_remove_batch());
-    EXPECT_GT(after, stats.get_last_remove_batch());
+    EXPECT_EQ(1, listener->remove_batch_cnt);
 }
 
 }