Avoid flushing components in proton more than necessary.

Don't consider TLSSIZE ordering if there exists an active (ongoing) flush (for the same flush handler)
that started before the last flush time of the flush target to evaluate.
Instead we should wait for the active (ongoing) flush to be finished before doing another evaluation.

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));
     }
 }