Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[fix](memory) Refactor MemCounter #40542

Merged
merged 4 commits into from
Sep 16, 2024
Merged

[fix](memory) Refactor MemCounter #40542

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
35c38b2
1
xinyiZzz Sep 13, 2024
29192ca
2
xinyiZzz Sep 13, 2024
867f395
3
xinyiZzz Sep 13, 2024
6bf216f
5
xinyiZzz Sep 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion be/src/cloud/cloud_tablet_mgr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ class CloudTabletMgr::TabletMap {
CloudTabletMgr::CloudTabletMgr(CloudStorageEngine& engine)
: _engine(engine),
_tablet_map(std::make_unique<TabletMap>()),
_cache(std::make_unique<LRUCachePolicyTrackingManual>(
_cache(std::make_unique<LRUCachePolicy>(
CachePolicy::CacheType::CLOUD_TABLET_CACHE, config::tablet_cache_capacity,
LRUCacheType::NUMBER, 0, config::tablet_cache_shards)) {}

Expand Down
4 changes: 2 additions & 2 deletions be/src/cloud/cloud_txn_delete_bitmap_cache.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@
namespace doris {

CloudTxnDeleteBitmapCache::CloudTxnDeleteBitmapCache(size_t size_in_bytes)
: LRUCachePolicyTrackingManual(CachePolicy::CacheType::CLOUD_TXN_DELETE_BITMAP_CACHE,
size_in_bytes, LRUCacheType::SIZE, 86400, 4),
: LRUCachePolicy(CachePolicy::CacheType::CLOUD_TXN_DELETE_BITMAP_CACHE, size_in_bytes,
LRUCacheType::SIZE, 86400, 4),
_stop_latch(1) {}

CloudTxnDeleteBitmapCache::~CloudTxnDeleteBitmapCache() {
Expand Down
2 changes: 1 addition & 1 deletion be/src/cloud/cloud_txn_delete_bitmap_cache.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
namespace doris {

// Record transaction related delete bitmaps using a lru cache.
class CloudTxnDeleteBitmapCache : public LRUCachePolicyTrackingManual {
class CloudTxnDeleteBitmapCache : public LRUCachePolicy {
public:
CloudTxnDeleteBitmapCache(size_t size_in_bytes);

Expand Down
9 changes: 4 additions & 5 deletions be/src/http/default_path_handlers.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,7 @@ void display_tablets_callback(const WebPageHandler::ArgumentMap& args, EasyJson*
// Registered to handle "/mem_tracker", and prints out memory tracker information.
void mem_tracker_handler(const WebPageHandler::ArgumentMap& args, std::stringstream* output) {
(*output) << "<h1>Memory usage by subsystem</h1>\n";
std::vector<MemTracker::Snapshot> snapshots;
std::vector<MemTrackerLimiter::Snapshot> snapshots;
auto iter = args.find("type");
if (iter != args.end()) {
if (iter->second == "global") {
Expand All @@ -159,7 +159,7 @@ void mem_tracker_handler(const WebPageHandler::ArgumentMap& args, std::stringstr
} else if (iter->second == "other") {
MemTrackerLimiter::make_type_snapshots(&snapshots, MemTrackerLimiter::Type::OTHER);
} else if (iter->second == "reserved_memory") {
GlobalMemoryArbitrator::make_reserved_memory_snapshots(&snapshots);
MemTrackerLimiter::make_all_reserved_trackers_snapshots(&snapshots);
} else if (iter->second == "all") {
MemTrackerLimiter::make_all_memory_state_snapshots(&snapshots);
}
Expand Down Expand Up @@ -191,7 +191,6 @@ void mem_tracker_handler(const WebPageHandler::ArgumentMap& args, std::stringstr
(*output) << "<thead><tr>"
"<th data-sortable='true'>Type</th>"
"<th data-sortable='true'>Label</th>"
"<th data-sortable='true'>Parent Label</th>"
"<th>Limit</th>"
"<th data-sortable='true' "
">Current Consumption(Bytes)</th>"
Expand All @@ -207,8 +206,8 @@ void mem_tracker_handler(const WebPageHandler::ArgumentMap& args, std::stringstr
string peak_consumption_normalize = AccurateItoaKMGT(item.peak_consumption);
(*output) << strings::Substitute(
"<tr><td>$0</td><td>$1</td><td>$2</td><td>$3</td><td>$4</td><td>$5</td><td>$6</"
"td><td>$7</td></tr>\n",
item.type, item.label, item.parent_label, limit_str, item.cur_consumption,
"td></tr>\n",
item.type, item.label, limit_str, item.cur_consumption,
current_consumption_normalize, item.peak_consumption, peak_consumption_normalize);
}
(*output) << "</tbody></table>\n";
Expand Down
5 changes: 2 additions & 3 deletions be/src/olap/memtable.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,9 +161,8 @@ MemTable::~MemTable() {
std::for_each(_row_in_blocks.begin(), _row_in_blocks.end(), std::default_delete<RowInBlock>());
_insert_mem_tracker->release(_mem_usage);
_flush_mem_tracker->set_consumption(0);
DCHECK_EQ(_insert_mem_tracker->consumption(), 0)
<< std::endl
<< MemTracker::log_usage(_insert_mem_tracker->make_snapshot());
DCHECK_EQ(_insert_mem_tracker->consumption(), 0) << std::endl
<< _insert_mem_tracker->log_usage();
DCHECK_EQ(_flush_mem_tracker->consumption(), 0);
_arena.reset();
_agg_buffer_pool.clear();
Expand Down
5 changes: 1 addition & 4 deletions be/src/olap/memtable_memory_limiter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,10 +62,7 @@ Status MemTableMemoryLimiter::init(int64_t process_mem_limit) {
_load_hard_mem_limit * config::load_process_safe_mem_permit_percent / 100;
g_load_hard_mem_limit.set_value(_load_hard_mem_limit);
g_load_soft_mem_limit.set_value(_load_soft_mem_limit);
_memtable_tracker_set =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::LOAD, "MemTableTrackerSet");
_mem_tracker = std::make_unique<MemTracker>("AllMemTableMemory",
ExecEnv::GetInstance()->details_mem_tracker_set());
_mem_tracker = std::make_unique<MemTracker>("AllMemTableMemory");
REGISTER_HOOK_METRIC(memtable_memory_limiter_mem_consumption,
[this]() { return _mem_tracker->consumption(); });
_log_timer.start();
Expand Down
5 changes: 1 addition & 4 deletions be/src/olap/memtable_memory_limiter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
#include <stdint.h>

#include "common/status.h"
#include "runtime/memory/mem_tracker_limiter.h"
#include "runtime/memory/mem_tracker.h"
#include "util/countdown_latch.h"
#include "util/stopwatch.hpp"

Expand All @@ -45,7 +45,6 @@ class MemTableMemoryLimiter {

void refresh_mem_tracker();

MemTrackerLimiter* memtable_tracker_set() { return _memtable_tracker_set.get(); }
MemTracker* mem_tracker() { return _mem_tracker.get(); }

int64_t mem_usage() const { return _mem_usage; }
Expand All @@ -68,8 +67,6 @@ class MemTableMemoryLimiter {
int64_t _write_mem_usage = 0;
int64_t _active_mem_usage = 0;

// mem tracker collection of all mem tables.
std::shared_ptr<MemTrackerLimiter> _memtable_tracker_set;
// sum of all mem table memory.
std::unique_ptr<MemTracker> _mem_tracker;
int64_t _load_hard_mem_limit = -1;
Expand Down
13 changes: 0 additions & 13 deletions be/src/olap/memtable_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,25 +187,12 @@ Status MemTableWriter::wait_flush() {
}

void MemTableWriter::_reset_mem_table() {
#ifndef BE_TEST
auto mem_table_insert_tracker = std::make_shared<MemTracker>(
fmt::format("MemTableManualInsert:TabletId={}:MemTableNum={}#loadID={}",
std::to_string(tablet_id()), _mem_table_num,
UniqueId(_req.load_id).to_string()),
ExecEnv::GetInstance()->memtable_memory_limiter()->memtable_tracker_set());
auto mem_table_flush_tracker = std::make_shared<MemTracker>(
fmt::format("MemTableHookFlush:TabletId={}:MemTableNum={}#loadID={}",
std::to_string(tablet_id()), _mem_table_num++,
UniqueId(_req.load_id).to_string()),
ExecEnv::GetInstance()->memtable_memory_limiter()->memtable_tracker_set());
#else
auto mem_table_insert_tracker = std::make_shared<MemTracker>(fmt::format(
"MemTableManualInsert:TabletId={}:MemTableNum={}#loadID={}",
std::to_string(tablet_id()), _mem_table_num, UniqueId(_req.load_id).to_string()));
auto mem_table_flush_tracker = std::make_shared<MemTracker>(fmt::format(
"MemTableHookFlush:TabletId={}:MemTableNum={}#loadID={}", std::to_string(tablet_id()),
_mem_table_num++, UniqueId(_req.load_id).to_string()));
#endif
{
std::lock_guard<SpinLock> l(_mem_table_tracker_lock);
_mem_table_insert_trackers.push_back(mem_table_insert_tracker);
Expand Down
26 changes: 13 additions & 13 deletions be/src/olap/page_cache.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,28 +92,28 @@ class StoragePageCache {
}
};

class DataPageCache : public LRUCachePolicyTrackingAllocator {
class DataPageCache : public LRUCachePolicy {
public:
DataPageCache(size_t capacity, uint32_t num_shards)
: LRUCachePolicyTrackingAllocator(
CachePolicy::CacheType::DATA_PAGE_CACHE, capacity, LRUCacheType::SIZE,
config::data_page_cache_stale_sweep_time_sec, num_shards) {}
: LRUCachePolicy(CachePolicy::CacheType::DATA_PAGE_CACHE, capacity,
LRUCacheType::SIZE, config::data_page_cache_stale_sweep_time_sec,
num_shards) {}
};

class IndexPageCache : public LRUCachePolicyTrackingAllocator {
class IndexPageCache : public LRUCachePolicy {
public:
IndexPageCache(size_t capacity, uint32_t num_shards)
: LRUCachePolicyTrackingAllocator(
CachePolicy::CacheType::INDEXPAGE_CACHE, capacity, LRUCacheType::SIZE,
config::index_page_cache_stale_sweep_time_sec, num_shards) {}
: LRUCachePolicy(CachePolicy::CacheType::INDEXPAGE_CACHE, capacity,
LRUCacheType::SIZE, config::index_page_cache_stale_sweep_time_sec,
num_shards) {}
};

class PKIndexPageCache : public LRUCachePolicyTrackingAllocator {
class PKIndexPageCache : public LRUCachePolicy {
public:
PKIndexPageCache(size_t capacity, uint32_t num_shards)
: LRUCachePolicyTrackingAllocator(
CachePolicy::CacheType::PK_INDEX_PAGE_CACHE, capacity, LRUCacheType::SIZE,
config::pk_index_page_cache_stale_sweep_time_sec, num_shards) {}
: LRUCachePolicy(CachePolicy::CacheType::PK_INDEX_PAGE_CACHE, capacity,
LRUCacheType::SIZE,
config::pk_index_page_cache_stale_sweep_time_sec, num_shards) {}
};

static constexpr uint32_t kDefaultNumShards = 16;
Expand Down Expand Up @@ -164,7 +164,7 @@ class StoragePageCache {
// delete bitmap in unique key with mow
std::unique_ptr<PKIndexPageCache> _pk_index_page_cache;

LRUCachePolicyTrackingAllocator* _get_page_cache(segment_v2::PageTypePB page_type) {
LRUCachePolicy* _get_page_cache(segment_v2::PageTypePB page_type) {
switch (page_type) {
case segment_v2::DATA_PAGE: {
return _data_page_cache.get();
Expand Down
6 changes: 3 additions & 3 deletions be/src/olap/rowset/segment_v2/inverted_index_cache.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,9 +135,9 @@ void InvertedIndexQueryCache::insert(const CacheKey& key, std::shared_ptr<roarin
return;
}

auto* lru_handle = LRUCachePolicyTrackingManual::insert(
key.encode(), (void*)cache_value_ptr.release(), bitmap->getSizeInBytes(),
bitmap->getSizeInBytes(), CachePriority::NORMAL);
auto* lru_handle = LRUCachePolicy::insert(key.encode(), (void*)cache_value_ptr.release(),
bitmap->getSizeInBytes(), bitmap->getSizeInBytes(),
CachePriority::NORMAL);
*handle = InvertedIndexQueryCacheHandle(this, lru_handle);
}

Expand Down
31 changes: 15 additions & 16 deletions be/src/olap/rowset/segment_v2/inverted_index_cache.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,23 +99,23 @@ class InvertedIndexSearcherCache {
private:
InvertedIndexSearcherCache() = default;

class InvertedIndexSearcherCachePolicy : public LRUCachePolicyTrackingManual {
class InvertedIndexSearcherCachePolicy : public LRUCachePolicy {
public:
InvertedIndexSearcherCachePolicy(size_t capacity, uint32_t num_shards,
uint32_t element_count_capacity)
: LRUCachePolicyTrackingManual(CachePolicy::CacheType::INVERTEDINDEX_SEARCHER_CACHE,
capacity, LRUCacheType::SIZE,
config::inverted_index_cache_stale_sweep_time_sec,
num_shards, element_count_capacity, true) {}
: LRUCachePolicy(CachePolicy::CacheType::INVERTEDINDEX_SEARCHER_CACHE, capacity,
LRUCacheType::SIZE,
config::inverted_index_cache_stale_sweep_time_sec, num_shards,
element_count_capacity, true) {}
InvertedIndexSearcherCachePolicy(size_t capacity, uint32_t num_shards,
uint32_t element_count_capacity,
CacheValueTimeExtractor cache_value_time_extractor,
bool cache_value_check_timestamp)
: LRUCachePolicyTrackingManual(
CachePolicy::CacheType::INVERTEDINDEX_SEARCHER_CACHE, capacity,
LRUCacheType::SIZE, config::inverted_index_cache_stale_sweep_time_sec,
num_shards, element_count_capacity, cache_value_time_extractor,
cache_value_check_timestamp, true) {}
: LRUCachePolicy(CachePolicy::CacheType::INVERTEDINDEX_SEARCHER_CACHE, capacity,
LRUCacheType::SIZE,
config::inverted_index_cache_stale_sweep_time_sec, num_shards,
element_count_capacity, cache_value_time_extractor,
cache_value_check_timestamp, true) {}
};
// Insert a cache entry by key.
// And the cache entry will be returned in handle.
Expand Down Expand Up @@ -179,9 +179,9 @@ class InvertedIndexCacheHandle {

class InvertedIndexQueryCacheHandle;

class InvertedIndexQueryCache : public LRUCachePolicyTrackingManual {
class InvertedIndexQueryCache : public LRUCachePolicy {
public:
using LRUCachePolicyTrackingManual::insert;
using LRUCachePolicy::insert;

// cache key
struct CacheKey {
Expand Down Expand Up @@ -227,10 +227,9 @@ class InvertedIndexQueryCache : public LRUCachePolicyTrackingManual {
InvertedIndexQueryCache() = delete;

InvertedIndexQueryCache(size_t capacity, uint32_t num_shards)
: LRUCachePolicyTrackingManual(CachePolicy::CacheType::INVERTEDINDEX_QUERY_CACHE,
capacity, LRUCacheType::SIZE,
config::inverted_index_cache_stale_sweep_time_sec,
num_shards) {}
: LRUCachePolicy(CachePolicy::CacheType::INVERTEDINDEX_QUERY_CACHE, capacity,
LRUCacheType::SIZE, config::inverted_index_cache_stale_sweep_time_sec,
num_shards) {}

bool lookup(const CacheKey& key, InvertedIndexQueryCacheHandle* handle);

Expand Down
7 changes: 3 additions & 4 deletions be/src/olap/schema_cache.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ using SegmentIteratorUPtr = std::unique_ptr<SegmentIterator>;
// eliminating the need for frequent allocation and deallocation during usage.
// This caching mechanism proves immensely advantageous, particularly in scenarios
// with high concurrency, where queries are executed simultaneously.
class SchemaCache : public LRUCachePolicyTrackingManual {
class SchemaCache : public LRUCachePolicy {
public:
static SchemaCache* instance();

Expand Down Expand Up @@ -86,9 +86,8 @@ class SchemaCache : public LRUCachePolicyTrackingManual {
};

SchemaCache(size_t capacity)
: LRUCachePolicyTrackingManual(CachePolicy::CacheType::SCHEMA_CACHE, capacity,
LRUCacheType::NUMBER,
config::schema_cache_sweep_time_sec) {}
: LRUCachePolicy(CachePolicy::CacheType::SCHEMA_CACHE, capacity, LRUCacheType::NUMBER,
config::schema_cache_sweep_time_sec) {}

private:
static constexpr char SCHEMA_DELIMITER = '-';
Expand Down
6 changes: 3 additions & 3 deletions be/src/olap/segment_loader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,9 +40,9 @@ bool SegmentCache::lookup(const SegmentCache::CacheKey& key, SegmentCacheHandle*

void SegmentCache::insert(const SegmentCache::CacheKey& key, SegmentCache::CacheValue& value,
SegmentCacheHandle* handle) {
auto* lru_handle = LRUCachePolicyTrackingManual::insert(
key.encode(), &value, value.segment->meta_mem_usage(), value.segment->meta_mem_usage(),
CachePriority::NORMAL);
auto* lru_handle =
LRUCachePolicy::insert(key.encode(), &value, value.segment->meta_mem_usage(),
value.segment->meta_mem_usage(), CachePriority::NORMAL);
handle->push_segment(this, lru_handle);
}

Expand Down
11 changes: 5 additions & 6 deletions be/src/olap/segment_loader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,9 +55,9 @@ class BetaRowset;
// Make sure that cache_handle is valid during the segment usage period.
using BetaRowsetSharedPtr = std::shared_ptr<BetaRowset>;

class SegmentCache : public LRUCachePolicyTrackingManual {
class SegmentCache : public LRUCachePolicy {
public:
using LRUCachePolicyTrackingManual::insert;
using LRUCachePolicy::insert;
// The cache key or segment lru cache
struct CacheKey {
CacheKey(RowsetId rowset_id_, int64_t segment_id_)
Expand All @@ -81,10 +81,9 @@ class SegmentCache : public LRUCachePolicyTrackingManual {
};

SegmentCache(size_t memory_bytes_limit, size_t segment_num_limit)
: LRUCachePolicyTrackingManual(CachePolicy::CacheType::SEGMENT_CACHE,
memory_bytes_limit, LRUCacheType::SIZE,
config::tablet_rowset_stale_sweep_time_sec,
DEFAULT_LRU_CACHE_NUM_SHARDS * 2, segment_num_limit) {}
: LRUCachePolicy(CachePolicy::CacheType::SEGMENT_CACHE, memory_bytes_limit,
LRUCacheType::SIZE, config::tablet_rowset_stale_sweep_time_sec,
DEFAULT_LRU_CACHE_NUM_SHARDS * 2, segment_num_limit) {}

// Lookup the given segment in the cache.
// If the segment is found, the cache entry will be written into handle.
Expand Down
8 changes: 4 additions & 4 deletions be/src/olap/storage_engine.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -540,7 +540,7 @@ class StorageEngine final : public BaseStorageEngine {
// lru cache for create tabelt round robin in disks
// key: partitionId_medium
// value: index
class CreateTabletIdxCache : public LRUCachePolicyTrackingManual {
class CreateTabletIdxCache : public LRUCachePolicy {
public:
// get key, delimiter with DELIMITER '-'
static std::string get_key(int64_t partition_id, TStorageMedium::type medium) {
Expand All @@ -558,9 +558,9 @@ class CreateTabletIdxCache : public LRUCachePolicyTrackingManual {
};

CreateTabletIdxCache(size_t capacity)
: LRUCachePolicyTrackingManual(CachePolicy::CacheType::CREATE_TABLET_RR_IDX_CACHE,
capacity, LRUCacheType::NUMBER,
/*stale_sweep_time_s*/ 30 * 60) {}
: LRUCachePolicy(CachePolicy::CacheType::CREATE_TABLET_RR_IDX_CACHE, capacity,
LRUCacheType::NUMBER,
/*stale_sweep_time_s*/ 30 * 60) {}
};

struct DirInfo {
Expand Down
3 changes: 1 addition & 2 deletions be/src/olap/tablet_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,8 +90,7 @@ bvar::Adder<int64_t> g_tablet_meta_schema_columns_count("tablet_meta_schema_colu

TabletManager::TabletManager(StorageEngine& engine, int32_t tablet_map_lock_shard_size)
: _engine(engine),
_tablet_meta_mem_tracker(std::make_shared<MemTracker>(
"TabletMeta(experimental)", ExecEnv::GetInstance()->details_mem_tracker_set())),
_tablet_meta_mem_tracker(std::make_shared<MemTracker>("TabletMeta(experimental)")),
_tablets_shards_size(tablet_map_lock_shard_size),
_tablets_shards_mask(tablet_map_lock_shard_size - 1) {
CHECK_GT(_tablets_shards_size, 0);
Expand Down
9 changes: 4 additions & 5 deletions be/src/olap/tablet_meta.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,13 +520,12 @@ class DeleteBitmap {

void remove_sentinel_marks();

class AggCachePolicy : public LRUCachePolicyTrackingManual {
class AggCachePolicy : public LRUCachePolicy {
public:
AggCachePolicy(size_t capacity)
: LRUCachePolicyTrackingManual(CachePolicy::CacheType::DELETE_BITMAP_AGG_CACHE,
capacity, LRUCacheType::SIZE,
config::delete_bitmap_agg_cache_stale_sweep_time_sec,
256) {}
: LRUCachePolicy(CachePolicy::CacheType::DELETE_BITMAP_AGG_CACHE, capacity,
LRUCacheType::SIZE,
config::delete_bitmap_agg_cache_stale_sweep_time_sec, 256) {}
};

class AggCache {
Expand Down
Loading
Loading