Merge pull request #25282 from vespa-engine/geirst/avoid-flushing-components-more-than-necessary

Avoid flushing components in proton more than necessary.

3 files changed, 93 insertions, 50 deletions

diff --git a/searchcore/src/tests/proton/flushengine/flushengine_test.cpp b/searchcore/src/tests/proton/flushengine/flushengine_test.cpp
index 4595ce12229..d795ed95550 100644
--- a/searchcore/src/tests/proton/flushengine/flushengine_test.cpp
+++ b/searchcore/src/tests/proton/flushengine/flushengine_test.cpp
@@ -1,5 +1,6 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
+#include <vespa/searchcore/proton/flushengine/active_flush_stats.h>
 #include <vespa/searchcore/proton/flushengine/cachedflushtarget.h>
 #include <vespa/searchcore/proton/flushengine/flush_engine_explorer.h>
 #include <vespa/searchcore/proton/flushengine/flushengine.h>
@@ -19,13 +20,8 @@
 #include <vespa/log/log.h>
 LOG_SETUP("flushengine_test");
 
-// --------------------------------------------------------------------------------
-//
-// Setup.
-//
-// --------------------------------------------------------------------------------
-
 using namespace proton;
+using namespace proton::flushengine;
 using namespace vespalib::slime;
 using searchcorespi::IFlushTarget;
 using searchcorespi::FlushTask;
@@ -335,8 +331,9 @@ public:
         const SimpleStrategy &_flush;
     };
 
-    FlushContext::List getFlushTargets(const FlushContext::List &targetList,
-                                       const flushengine::TlsStatsMap &) const override {
+    FlushContext::List getFlushTargets(const FlushContext::List& targetList,
+                                       const flushengine::TlsStatsMap&,
+                                       const flushengine::ActiveFlushStats&) const override {
         FlushContext::List fv(targetList);
         std::sort(fv.begin(), fv.end(), CompareTarget(*this));
         return fv;
@@ -380,8 +377,8 @@ public:
 
 class NoFlushStrategy : public SimpleStrategy
 {
-    FlushContext::List getFlushTargets(const FlushContext::List &, const flushengine::TlsStatsMap &) const override {
-        return FlushContext::List();
+    FlushContext::List getFlushTargets(const FlushContext::List &, const flushengine::TlsStatsMap &, const flushengine::ActiveFlushStats&) const override {
+        return {};
     }
 };
 
@@ -771,6 +768,28 @@ TEST_F("require that oldest serial is updated when finishing priority flush stra
     EXPECT_EQUAL(20u, handler->_oldestSerial);
 }
 
+TEST("the oldest start time is tracked per flush handler in ActiveFlushStats")
+{
+    using seconds = std::chrono::seconds;
+    using vespalib::system_time;
+    system_time now = vespalib::system_clock::now();
+    system_time t1 = now + seconds(1);
+    system_time t2 = now + seconds(2);
+    system_time t3 = now + seconds(3);
+    system_time t4 = now + seconds(4);
+    ActiveFlushStats stats;
+    EXPECT_FALSE(stats.oldest_start_time("h1").has_value());
+    stats.set_start_time("h1", t2);
+    stats.set_start_time("h2", t4);
+    EXPECT_EQUAL(t2, stats.oldest_start_time("h1").value());
+    EXPECT_EQUAL(t4, stats.oldest_start_time("h2").value());
+
+    stats.set_start_time("h1", t1);
+    EXPECT_EQUAL(t1, stats.oldest_start_time("h1").value());
+    stats.set_start_time("h1", t3);
+    EXPECT_EQUAL(t1, stats.oldest_start_time("h1").value());
+}
+
 
 TEST_MAIN()
 {
diff --git a/searchcore/src/tests/proton/flushengine/prepare_restart_flush_strategy/prepare_restart_flush_strategy_test.cpp b/searchcore/src/tests/proton/flushengine/prepare_restart_flush_strategy/prepare_restart_flush_strategy_test.cpp
index 40c4c491940..6aa17151942 100644
--- a/searchcore/src/tests/proton/flushengine/prepare_restart_flush_strategy/prepare_restart_flush_strategy_test.cpp
+++ b/searchcore/src/tests/proton/flushengine/prepare_restart_flush_strategy/prepare_restart_flush_strategy_test.cpp
@@ -1,9 +1,10 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #include <vespa/vespalib/testkit/testapp.h>
 
-#include <vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h>
-#include <vespa/searchcore/proton/flushengine/flush_target_candidates.h>
+#include <vespa/searchcore/proton/flushengine/active_flush_stats.h>
 #include <vespa/searchcore/proton/flushengine/flush_target_candidate.h>
+#include <vespa/searchcore/proton/flushengine/flush_target_candidates.h>
+#include <vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h>
 #include <vespa/searchcore/proton/flushengine/tls_stats_map.h>
 #include <vespa/searchcore/proton/test/dummy_flush_handler.h>
 #include <vespa/searchcore/proton/test/dummy_flush_target.h>
@@ -220,7 +221,8 @@ struct FlushStrategyFixture
     {}
     FlushContext::List getFlushTargets(const FlushContext::List &targetList,
                                        const flushengine::TlsStatsMap &tlsStatsMap) const {
-        return strategy.getFlushTargets(targetList, tlsStatsMap);
+        flushengine::ActiveFlushStats active_flushes;
+        return strategy.getFlushTargets(targetList, tlsStatsMap, active_flushes);
     }
 };
 
diff --git a/searchcore/src/tests/proton/server/memoryflush/memoryflush_test.cpp b/searchcore/src/tests/proton/server/memoryflush/memoryflush_test.cpp
index 8a9b119bcbb..8c017184a7b 100644
--- a/searchcore/src/tests/proton/server/memoryflush/memoryflush_test.cpp
+++ b/searchcore/src/tests/proton/server/memoryflush/memoryflush_test.cpp
@@ -1,5 +1,6 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
+#include <vespa/searchcore/proton/flushengine/active_flush_stats.h>
 #include <vespa/searchcore/proton/flushengine/flushcontext.h>
 #include <vespa/searchcore/proton/flushengine/tls_stats_map.h>
 #include <vespa/searchcore/proton/server/memoryflush.h>
@@ -52,19 +53,15 @@ public:
     bool needUrgentFlush() const override { return _urgentFlush; }
 };
 
-struct StringList : public std::vector<vespalib::string> {
-    StringList() : std::vector<vespalib::string>() {}
-    StringList &add(const vespalib::string &str) {
-        push_back(str);
-        return *this;
-    }
-};
+using StringList = std::vector<vespalib::string>;
 
 class ContextBuilder {
 private:
     FlushContext::List _list;
     IFlushHandler::SP  _handler;
     flushengine::TlsStatsMap::Map _map;
+    flushengine::ActiveFlushStats _active_flushes;
+
     void
     fixupMap(const vespalib::string &name, SerialNum lastSerial)
     {
@@ -92,25 +89,32 @@ public:
         FlushContext::SP ctx(new FlushContext(_handler, target, lastSerial));
         return add(ctx);
     }
+    ContextBuilder& active_flush(const vespalib::string& handler_name, vespalib::system_time start_time) {
+        _active_flushes.set_start_time(handler_name, start_time);
+        return *this;
+    }
     const FlushContext::List &list() const { return _list; }
     flushengine::TlsStatsMap tlsStats() const {
         flushengine::TlsStatsMap::Map map(_map);
         return flushengine::TlsStatsMap(std::move(map));
     }
+    FlushContext::List flush_targets(const IFlushStrategy& strategy) const {
+        return strategy.getFlushTargets(list(), tlsStats(), _active_flushes);
+    }
 };
 
 using minutes = std::chrono::minutes;
 using seconds = std::chrono::seconds;
 
 ContextBuilder::ContextBuilder()
-    : _list(), _handler(new MyFlushHandler("myhandler"))
+    : _list(), _handler(new MyFlushHandler("myhandler")), _map(), _active_flushes()
 {}
 ContextBuilder::~ContextBuilder() = default;
 
 MyFlushTarget::SP
 createTargetM(const vespalib::string &name, MemoryGain memoryGain)
 {
-    return std::make_shared<MyFlushTarget>(name, memoryGain, DiskGain(),SerialNum(), system_time(), false);
+    return std::make_shared<MyFlushTarget>(name, memoryGain, DiskGain(), SerialNum(), system_time(), false);
 }
 
 MyFlushTarget::SP
@@ -155,13 +159,11 @@ TEST(MemoryFlushTest, can_order_by_memory_gain)
       .add(createTargetM("t3", MemoryGain(15, 0)));
     { // target t4 has memoryGain >= maxMemoryGain
         MemoryFlush flush({1000, 20_Gi, 1.0, 20, 1.0, minutes(1)});
-        assertOrder(StringList().add("t4").add("t3").add("t2").add("t1"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t3", "t2", "t1"}, cb.flush_targets(flush));
     }
     { // trigger totalMemoryGain >= globalMaxMemory
         MemoryFlush flush({50, 20_Gi, 1.0, 1000, 1.0, minutes(1)});
-        assertOrder(StringList().add("t4").add("t3").add("t2").add("t1"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t3", "t2", "t1"}, cb.flush_targets(flush));
     }
 }
 
@@ -178,14 +180,12 @@ TEST(MemoryFlushTest, can_order_by_disk_gain_with_large_values)
     { // target t4 has diskGain > bloatValue
         // t4 gain: 55M / 100M = 0.55 -> bloat factor 0.54 to trigger
         MemoryFlush flush({1000, 20_Gi, 10.0, 1000, 0.54, minutes(1)});
-        assertOrder(StringList().add("t4").add("t3").add("t2").add("t1"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t3", "t2", "t1"}, cb.flush_targets(flush));
     }
     { // trigger totalDiskGain > totalBloatValue
         // total gain: 160M / 4 * 100M = 0.4 -> bloat factor 0.39 to trigger
         MemoryFlush flush({1000, 20_Gi, 0.39, 1000, 10.0, minutes(1)});
-        assertOrder(StringList().add("t4").add("t3").add("t2").add("t1"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t3", "t2", "t1"}, cb.flush_targets(flush));
     }
 }
 
@@ -201,14 +201,12 @@ TEST(MemoryFlushTest, can_order_by_disk_gain_with_small_values)
     { // target t4 has diskGain > bloatValue
         // t4 gain: 55 / 100M = 0.0000055 -> bloat factor 0.0000054 to trigger
         MemoryFlush flush({1000, 20_Gi, 10.0, 1000, 0.00000054, minutes(1)});
-        assertOrder(StringList().add("t4").add("t3").add("t2").add("t1"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t3", "t2", "t1"}, cb.flush_targets(flush));
     }
     { // trigger totalDiskGain > totalBloatValue
         // total gain: 160 / 100M = 0.0000016 -> bloat factor 0.0000015 to trigger
         MemoryFlush flush({1000, 20_Gi, 0.0000015, 1000, 10.0, minutes(1)});
-        assertOrder(StringList().add("t4").add("t3").add("t2").add("t1"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t3", "t2", "t1"}, cb.flush_targets(flush));
     }
 }
 
@@ -224,12 +222,11 @@ TEST(MemoryFlushTest, can_order_by_age)
 
     { // all targets have timeDiff >= maxTimeGain
         MemoryFlush flush({1000, 20_Gi, 1.0, 1000, 1.0, seconds(2)}, start);
-        assertOrder(StringList().add("t4").add("t3").add("t2").add("t1"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t3", "t2", "t1"}, cb.flush_targets(flush));
     }
     { // no targets have timeDiff >= maxTimeGain
         MemoryFlush flush({1000, 20_Gi, 1.0, 1000, 1.0, seconds(30)}, start);
-        assertOrder(StringList(), flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({}, cb.flush_targets(flush));
     }
 }
 
@@ -238,8 +235,8 @@ TEST(MemoryFlushTest, can_order_by_tls_size)
     system_time now(vespalib::system_clock::now());
     system_time start = now - seconds(20);
     ContextBuilder cb;
-    IFlushHandler::SP handler1(std::make_shared<MyFlushHandler>("handler1"));
-    IFlushHandler::SP handler2(std::make_shared<MyFlushHandler>("handler2"));
+    auto handler1 = std::make_shared<MyFlushHandler>("handler1");
+    auto handler2 = std::make_shared<MyFlushHandler>("handler2");
     cb.addTls("handler1", {20_Gi, 1001, 2000 });
     cb.addTls("handler2", { 5_Gi, 1001, 2000 });
     cb.add(std::make_shared<FlushContext>(handler1, createTargetT("t2", now - seconds(10), 1900), 2000)).
@@ -248,27 +245,52 @@ TEST(MemoryFlushTest, can_order_by_tls_size)
         add(std::make_shared<FlushContext>(handler2, createTargetT("t3", now - seconds(15), 1900), 2000));
     { // sum of tls sizes above limit, trigger sort order based on tls size
         MemoryFlush flush({1000, 3_Gi, 1.0, 1000, 1.0, seconds(2)}, start);
-        assertOrder(StringList().add("t4").add("t1").add("t2").add("t3"),
-                    flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t4", "t1", "t2", "t3"}, cb.flush_targets(flush));
     }
     { // sum of tls sizes below limit
         MemoryFlush flush({1000, 30_Gi, 1.0, 1000, 1.0, seconds(30)}, start);
-        assertOrder(StringList(), flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({}, cb.flush_targets(flush));
     }
 }
 
+TEST(MemoryFlushTest, order_by_tls_size_not_always_selected_when_active_flushes)
+{
+    system_time now = vespalib::system_clock::now();
+    system_time start = now - seconds(20);
+    ContextBuilder cb;
+    auto h1 = std::make_shared<MyFlushHandler>("h1");
+    constexpr uint64_t first_serial = 1001;
+    constexpr uint64_t last_serial = 2000;
+    cb.addTls("h1", {5_Gi, first_serial, last_serial});
+    cb.add(std::make_shared<FlushContext>(h1, createTargetT("t1", now - seconds(10), 1300), last_serial)).
+       add(std::make_shared<FlushContext>(h1, createTargetT("t2", now - seconds(5), 1500), last_serial));
+    MemoryFlush flush({1000, 4_Gi, 1.0, 1000, 1.0, seconds(60)}, start);
+
+    // Last flush time of both t1 and t2 are older than start of active flush -> triggers TLSSIZE.
+    cb.active_flush("h1", now - seconds(4));
+    assertOrder({"t1", "t2"}, cb.flush_targets(flush));
+
+    // Last flush time of only t1 is older than start of active flush -> triggers TLSSIZE.
+    cb.active_flush("h1", now - seconds(9));
+    assertOrder({"t1", "t2"}, cb.flush_targets(flush));
+
+    // Last flush time both t1 and t2 is newer than start of active flush -> don't use TLSSIZE.
+    cb.active_flush("h1", now - seconds(11));
+    assertOrder({}, cb.flush_targets(flush));
+}
+
 TEST(MemoryFlushTest, can_handle_large_serial_numbers_when_ordering_by_tls_size)
 {
     uint64_t uint32_max = std::numeric_limits<uint32_t>::max();
-    ContextBuilder builder;
+    ContextBuilder cb;
     SerialNum firstSerial = 10;
     SerialNum lastSerial = uint32_max + 10;
-    builder.addTls("myhandler", {uint32_max, firstSerial, lastSerial});
-    builder.add(createTargetT("t1", system_time(), uint32_max + 5), lastSerial);
-    builder.add(createTargetT("t2", system_time(), uint32_max - 5), lastSerial);
+    cb.addTls("myhandler", {uint32_max, firstSerial, lastSerial});
+    cb.add(createTargetT("t1", system_time(), uint32_max + 5), lastSerial);
+    cb.add(createTargetT("t2", system_time(), uint32_max - 5), lastSerial);
     uint64_t maxMemoryGain = 10;
     MemoryFlush flush({maxMemoryGain, 1000, 0, maxMemoryGain, 0, vespalib::duration(0)}, system_time());
-    assertOrder(StringList().add("t2").add("t1"), flush.getFlushTargets(builder.list(), builder.tlsStats()));
+    assertOrder({"t2", "t1"}, cb.flush_targets(flush));
 }
 
 TEST(MemoryFlushTest, order_type_is_preserved)
@@ -281,21 +303,21 @@ TEST(MemoryFlushTest, order_type_is_preserved)
         cb.add(createTargetT("t2", ts2, 5), 14)
           .add(createTargetD("t1", DiskGain(100 * milli, 80 * milli), 5));
         MemoryFlush flush({1000, 20_Gi, 1.0, 1000, 0.19, seconds(30)});
-        assertOrder(StringList().add("t1").add("t2"), flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t1", "t2"}, cb.flush_targets(flush));
     }
     { // DISKBLOAT VS MEMORY
         ContextBuilder cb;
         cb.add(createTargetD("t2", DiskGain(100 * milli, 80 * milli)))
           .add(createTargetM("t1", MemoryGain(100, 80)));
         MemoryFlush flush({1000, 20_Gi, 1.0, 20, 0.19, seconds(30)});
-        assertOrder(StringList().add("t1").add("t2"), flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t1", "t2"}, cb.flush_targets(flush));
     }
     { // urgent flush
         ContextBuilder cb;
         cb.add(createTargetF("t2", false))
           .add(createTargetF("t1", true));
         MemoryFlush flush({1000, 20_Gi, 1.0, 1000, 1.0, seconds(30)});
-        assertOrder(StringList().add("t1").add("t2"), flush.getFlushTargets(cb.list(), cb.tlsStats()));
+        assertOrder({"t1", "t2"}, cb.flush_targets(flush));
     }
 }