diff options
Diffstat (limited to 'searchcore')
19 files changed, 258 insertions, 92 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)); } } diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/CMakeLists.txt b/searchcore/src/vespa/searchcore/proton/flushengine/CMakeLists.txt index a0e69af5b1e..7f013796fdb 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/CMakeLists.txt +++ b/searchcore/src/vespa/searchcore/proton/flushengine/CMakeLists.txt @@ -1,6 +1,7 @@ # Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. vespa_add_library(searchcore_flushengine STATIC SOURCES + active_flush_stats.cpp cachedflushtarget.cpp shrink_lid_space_flush_target.cpp flush_all_strategy.cpp diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/active_flush_stats.cpp b/searchcore/src/vespa/searchcore/proton/flushengine/active_flush_stats.cpp new file mode 100644 index 00000000000..0cbbc5c2a0d --- /dev/null +++ b/searchcore/src/vespa/searchcore/proton/flushengine/active_flush_stats.cpp @@ -0,0 +1,40 @@ +// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include "active_flush_stats.h" +#include <vespa/vespalib/stllike/hash_map.hpp> +#include <vespa/vespalib/util/time.h> + +namespace proton::flushengine { + +ActiveFlushStats::ActiveFlushStats() + : _stats() +{ +} + +void +ActiveFlushStats::set_start_time(const vespalib::string& handler_name, vespalib::system_time start_time) +{ + auto itr = _stats.find(handler_name); + if (itr != _stats.end()) { + if (start_time < itr->second) { + itr->second = start_time; + } + } else { + _stats.insert(std::make_pair(handler_name, start_time)); + } +} + +ActiveFlushStats::OptionalTime +ActiveFlushStats::oldest_start_time(const vespalib::string& handler_name) const +{ + auto itr = _stats.find(handler_name); + if (itr != _stats.end()) { + return OptionalTime(itr->second); + } + return std::nullopt; +} + +} + +VESPALIB_HASH_MAP_INSTANTIATE(vespalib::string, vespalib::system_time); + diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/active_flush_stats.h b/searchcore/src/vespa/searchcore/proton/flushengine/active_flush_stats.h new file mode 100644 index 00000000000..97502291569 --- /dev/null +++ b/searchcore/src/vespa/searchcore/proton/flushengine/active_flush_stats.h @@ -0,0 +1,32 @@ +// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +#pragma once + +#include <vespa/vespalib/stllike/hash_map.h> +#include <vespa/vespalib/util/time.h> +#include <optional> + +namespace proton::flushengine { + +/** + * Tracks the oldest start time of active (ongoing) flushes in each flush handler. + */ +class ActiveFlushStats { +public: + using OptionalTime = std::optional<vespalib::system_time>; + +private: + using StatsMap = vespalib::hash_map<vespalib::string, vespalib::system_time>; + StatsMap _stats; + +public: + ActiveFlushStats(); + /** + * Set the start time for a flush in the given flush handler. + * A start time is only updated if it is older than the current oldest one. + */ + void set_start_time(const vespalib::string& handler_name, vespalib::system_time start_time); + OptionalTime oldest_start_time(const vespalib::string& handler_name) const; +}; + +} + diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.cpp b/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.cpp index cad5fb05c9d..446f57fae74 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.cpp +++ b/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.cpp @@ -36,10 +36,11 @@ FlushAllStrategy::FlushAllStrategy() FlushContext::List FlushAllStrategy::getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap &) const + const flushengine::TlsStatsMap&, + const flushengine::ActiveFlushStats&) const { if (targetList.empty()) { - return FlushContext::List(); + return {}; } FlushContext::List fv(targetList); std::sort(fv.begin(), fv.end(), CompareTarget()); diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.h b/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.h index ac5028b3ab0..9acf8e362e7 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.h +++ b/searchcore/src/vespa/searchcore/proton/flushengine/flush_all_strategy.h @@ -16,7 +16,8 @@ public: FlushContext::List getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap &) const override; + const flushengine::TlsStatsMap&, + const flushengine::ActiveFlushStats&) const override; }; } // namespace proton diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.cpp b/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.cpp index 6ee0ae8e6b9..156b0c70d63 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.cpp +++ b/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.cpp @@ -20,7 +20,13 @@ FlushContext::FlushContext( vespalib::string FlushContext::createName(const IFlushHandler & handler, const IFlushTarget & target) { - return (handler.getName() + "." + target.getName()); + return create_name(handler.getName(), target.getName()); +} + +vespalib::string +FlushContext::create_name(const vespalib::string& handler_name, + const vespalib::string& target_name) { + return (handler_name + "." + target_name); } FlushContext::~FlushContext() diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.h b/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.h index 22b3fce3e88..c13d9e042e1 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.h +++ b/searchcore/src/vespa/searchcore/proton/flushengine/flushcontext.h @@ -35,6 +35,12 @@ public: static vespalib::string createName(const IFlushHandler & handler, const IFlushTarget & target); /** + * Create a combined name of the handler name and the target name. + */ + static vespalib::string create_name(const vespalib::string& handler_name, + const vespalib::string& target_name); + + /** * Constructs a new instance of this class. * * @param handler The flush handler that contains the given target. diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.cpp b/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.cpp index c0a1bcbd4c3..4968cb6791b 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.cpp +++ b/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.cpp @@ -1,6 +1,7 @@ // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. #include "flushengine.h" +#include "active_flush_stats.h" #include "cachedflushtarget.h" #include "flush_all_strategy.h" #include "flushtask.h" @@ -56,15 +57,17 @@ VESPA_THREAD_STACK_TAG(flush_engine_executor) } -FlushEngine::FlushMeta::FlushMeta(const vespalib::string & name, uint32_t id) - : _name(name), +FlushEngine::FlushMeta::FlushMeta(const vespalib::string& handler_name, + const vespalib::string& target_name, uint32_t id) + : _name((handler_name.empty() && target_name.empty()) ? "" : FlushContext::create_name(handler_name, target_name)), + _handler_name(handler_name), _timer(), _id(id) { } FlushEngine::FlushMeta::~FlushMeta() = default; FlushEngine::FlushInfo::FlushInfo() - : FlushMeta("", 0), + : FlushMeta("", "", 0), _target() { } @@ -72,8 +75,8 @@ FlushEngine::FlushInfo::FlushInfo() FlushEngine::FlushInfo::~FlushInfo() = default; -FlushEngine::FlushInfo::FlushInfo(uint32_t taskId, const IFlushTarget::SP &target, const vespalib::string & destination) - : FlushMeta(destination, taskId), +FlushEngine::FlushInfo::FlushInfo(uint32_t taskId, const vespalib::string& handler_name, const IFlushTarget::SP& target) + : FlushMeta(handler_name, target->getName(), taskId), _target(target) { } @@ -260,17 +263,32 @@ FlushEngine::getTargetList(bool includeFlushingTargets) const return ret; } +namespace { + +flushengine::ActiveFlushStats +make_active_flushes(const FlushEngine::FlushMetaSet& flush_set) +{ + flushengine::ActiveFlushStats result; + for (const auto& elem : flush_set) { + result.set_start_time(elem.handler_name(), elem.getStart()); + } + return result; +} + +} + std::pair<FlushContext::List,bool> FlushEngine::getSortedTargetList() { - FlushContext::List unsortedTargets = getTargetList(false); - flushengine::TlsStatsMap tlsStatsMap(_tlsStatsFactory->create()); + auto unsortedTargets = getTargetList(false); + auto tlsStatsMap = _tlsStatsFactory->create(); + auto active_flushes = make_active_flushes(getCurrentlyFlushingSet()); std::lock_guard<std::mutex> strategyGuard(_strategyLock); std::pair<FlushContext::List, bool> ret; if (_priorityStrategy) { - ret = std::make_pair(_priorityStrategy->getFlushTargets(unsortedTargets, tlsStatsMap), true); + ret = std::make_pair(_priorityStrategy->getFlushTargets(unsortedTargets, tlsStatsMap, active_flushes), true); } else { - ret = std::make_pair(_strategy->getFlushTargets(unsortedTargets, tlsStatsMap), false); + ret = std::make_pair(_strategy->getFlushTargets(unsortedTargets, tlsStatsMap, active_flushes), false); } return ret; } @@ -427,7 +445,7 @@ FlushEngine::initFlush(const IFlushHandler::SP &handler, const IFlushTarget::SP std::lock_guard<std::mutex> guard(_lock); taskId = _taskId++; vespalib::string name(FlushContext::createName(*handler, *target)); - FlushInfo flush(taskId, target, name); + FlushInfo flush(taskId, handler->getName(), target); _flushing[taskId] = flush; } LOG(debug, "FlushEngine::initFlush(handler='%s', target='%s') => taskId='%d'", diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h b/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h index 632f4482654..f8036ad4a92 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h +++ b/searchcore/src/vespa/searchcore/proton/flushengine/flushengine.h @@ -23,31 +23,33 @@ class FlushEngine final : public FastOS_Runnable public: class FlushMeta { public: - FlushMeta(const vespalib::string & name, uint32_t id); + FlushMeta(const vespalib::string& handler_name, const vespalib::string& target_name, uint32_t id); ~FlushMeta(); const vespalib::string & getName() const { return _name; } + const vespalib::string& handler_name() const { return _handler_name; } vespalib::system_time getStart() const { return vespalib::system_clock::now() - std::chrono::duration_cast<vespalib::system_time::duration>(elapsed()); } vespalib::duration elapsed() const { return _timer.elapsed(); } uint32_t getId() const { return _id; } bool operator < (const FlushMeta & rhs) const { return _id < rhs._id; } private: vespalib::string _name; + vespalib::string _handler_name; vespalib::Timer _timer; uint32_t _id; }; - typedef std::set<FlushMeta> FlushMetaSet; + using FlushMetaSet = std::set<FlushMeta>; private: using IFlushTarget = searchcorespi::IFlushTarget; struct FlushInfo : public FlushMeta { FlushInfo(); - FlushInfo(uint32_t taskId, const IFlushTarget::SP &target, const vespalib::string &destination); + FlushInfo(uint32_t taskId, const vespalib::string& handler_name, const IFlushTarget::SP &target); ~FlushInfo(); IFlushTarget::SP _target; }; - typedef std::map<uint32_t, FlushInfo> FlushMap; - typedef HandlerMap<IFlushHandler> FlushHandlerMap; + using FlushMap = std::map<uint32_t, FlushInfo>; + using FlushHandlerMap = HandlerMap<IFlushHandler>; bool _closed; const uint32_t _maxConcurrent; const vespalib::duration _idleInterval; @@ -89,8 +91,8 @@ public: /** * Convenience typedefs. */ - typedef std::unique_ptr<FlushEngine> UP; - typedef std::shared_ptr<FlushEngine> SP; + using UP = std::unique_ptr<FlushEngine>; + using SP = std::shared_ptr<FlushEngine>; /** * Constructs a new instance of this class. diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/iflushstrategy.h b/searchcore/src/vespa/searchcore/proton/flushengine/iflushstrategy.h index 6b64870096d..2c76b71ad13 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/iflushstrategy.h +++ b/searchcore/src/vespa/searchcore/proton/flushengine/iflushstrategy.h @@ -6,7 +6,10 @@ namespace proton { -namespace flushengine { class TlsStatsMap; } +namespace flushengine { +class ActiveFlushStats; +class TlsStatsMap; +} /** * This class represents a strategy used by the FlushEngine to make decisions on @@ -25,11 +28,13 @@ public: * Takes an input of targets that are candidates for flush and returns * a list of targets sorted according to priority strategy. * @param targetList The list of possible flush targets. - * @param lastSerial is the last serialnumber known by flushengine. + * @param tlsStatsMap Statistics per domain in the TLS. A domain matches a flush handler. + * @parma active_flushes Statistics of active (ongoing) flushes per flush handler. * @return A prioritized list of targets to flush. */ - virtual FlushContext::List getFlushTargets(const FlushContext::List & targetList, - const flushengine::TlsStatsMap & tlsStatsMap) const = 0; + virtual FlushContext::List getFlushTargets(const FlushContext::List& targetList, + const flushengine::TlsStatsMap& tlsStatsMap, + const flushengine::ActiveFlushStats& active_flushes) const = 0; protected: IFlushStrategy() = default; }; diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.cpp b/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.cpp index 028758e3fa2..5afdf6b5e1a 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.cpp +++ b/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.cpp @@ -153,7 +153,8 @@ findBestTargetsToFlushPerHandler(const FlushContextsMap &flushContextsPerHandler FlushContext::List PrepareRestartFlushStrategy::getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap &tlsStatsMap) const + const flushengine::TlsStatsMap &tlsStatsMap, + const flushengine::ActiveFlushStats&) const { return flatten(findBestTargetsToFlushPerHandler( groupByFlushHandler(removeGCFlushTargets(targetList)), diff --git a/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h b/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h index 61336b185d4..8ca9a5c98f7 100644 --- a/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h +++ b/searchcore/src/vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h @@ -36,8 +36,8 @@ public: PrepareRestartFlushStrategy(const Config &cfg); virtual FlushContext::List getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap & - tlsStatsMap) const override; + const flushengine::TlsStatsMap &tlsStatsMap, + const flushengine::ActiveFlushStats&) const override; }; } // namespace proton diff --git a/searchcore/src/vespa/searchcore/proton/server/memoryflush.cpp b/searchcore/src/vespa/searchcore/proton/server/memoryflush.cpp index 31089233032..f8d7519fd0c 100644 --- a/searchcore/src/vespa/searchcore/proton/server/memoryflush.cpp +++ b/searchcore/src/vespa/searchcore/proton/server/memoryflush.cpp @@ -1,12 +1,12 @@ // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. #include "memoryflush.h" +#include <vespa/searchcore/proton/flushengine/active_flush_stats.h> #include <vespa/searchcore/proton/flushengine/tls_stats_map.h> #include <vespa/vespalib/stllike/asciistream.h> #include <vespa/vespalib/stllike/hash_set.h> -#include <vespa/vespalib/util/time.h> #include <vespa/vespalib/util/size_literals.h> -#include <algorithm> +#include <vespa/vespalib/util/time.h> #include <vespa/log/log.h> LOG_SETUP(".proton.server.memoryflush"); @@ -136,8 +136,9 @@ computeGain(const IFlushTarget::DiskGain & gain) { } FlushContext::List -MemoryFlush::getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap & tlsStatsMap) const +MemoryFlush::getFlushTargets(const FlushContext::List& targetList, + const flushengine::TlsStatsMap& tlsStatsMap, + const flushengine::ActiveFlushStats& active_flushes) const { OrderType order(DEFAULT); uint64_t totalMemory(0); @@ -166,10 +167,17 @@ MemoryFlush::getFlushTargets(const FlushContext::List &targetList, vespalib::duration timeDiff(now - (lastFlushTime > vespalib::system_time() ? lastFlushTime : _startTime)); totalMemory += mgain; const flushengine::TlsStats &tlsStats = tlsStatsMap.getTlsStats(handler.getName()); - if (visitedHandlers.insert(&handler).second) { - totalTlsSize += tlsStats.getNumBytes(); - if ((totalTlsSize > config.maxGlobalTlsSize) && (order < TLSSIZE)) { - order = TLSSIZE; + + auto oldest_start_time = active_flushes.oldest_start_time(handler.getName()); + // Don't consider TLSSIZE if there exists an active (ongoing) flush (for this 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. + if (!oldest_start_time.has_value() || lastFlushTime < oldest_start_time.value()) { + if (visitedHandlers.insert(&handler).second) { + totalTlsSize += tlsStats.getNumBytes(); + if ((totalTlsSize > config.maxGlobalTlsSize) && (order < TLSSIZE)) { + order = TLSSIZE; + } } } if ((mgain >= config.maxMemoryGain) && (order < MEMORY)) { diff --git a/searchcore/src/vespa/searchcore/proton/server/memoryflush.h b/searchcore/src/vespa/searchcore/proton/server/memoryflush.h index 02d2945ab4f..82c988bef9f 100644 --- a/searchcore/src/vespa/searchcore/proton/server/memoryflush.h +++ b/searchcore/src/vespa/searchcore/proton/server/memoryflush.h @@ -80,7 +80,8 @@ public: FlushContext::List getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap &tlsStatsMap) const override; + const flushengine::TlsStatsMap &tlsStatsMap, + const flushengine::ActiveFlushStats& active_flushes) const override; void setConfig(const Config &config); Config getConfig() const; diff --git a/searchcore/src/vespa/searchcore/proton/server/simpleflush.cpp b/searchcore/src/vespa/searchcore/proton/server/simpleflush.cpp index 2de6dadcd26..a1234ccc8fc 100644 --- a/searchcore/src/vespa/searchcore/proton/server/simpleflush.cpp +++ b/searchcore/src/vespa/searchcore/proton/server/simpleflush.cpp @@ -13,10 +13,10 @@ SimpleFlush::SimpleFlush() } FlushContext::List -SimpleFlush::getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap &tlsStatsMap) const +SimpleFlush::getFlushTargets(const FlushContext::List& targetList, + const flushengine::TlsStatsMap&, + const flushengine::ActiveFlushStats&) const { - (void) tlsStatsMap; FlushContext::List fv(targetList); std::sort(fv.begin(), fv.end(), CompareTarget()); return fv; diff --git a/searchcore/src/vespa/searchcore/proton/server/simpleflush.h b/searchcore/src/vespa/searchcore/proton/server/simpleflush.h index 3a066e100f0..1b6053842b1 100644 --- a/searchcore/src/vespa/searchcore/proton/server/simpleflush.h +++ b/searchcore/src/vespa/searchcore/proton/server/simpleflush.h @@ -21,8 +21,9 @@ public: SimpleFlush(); // Implements IFlushStrategy - virtual FlushContext::List getFlushTargets(const FlushContext::List &targetList, - const flushengine::TlsStatsMap &tlsStatsMap) const override; + virtual FlushContext::List getFlushTargets(const FlushContext::List& targetList, + const flushengine::TlsStatsMap&, + const flushengine::ActiveFlushStats&) const override; }; |