Skip to content

Add stats for grouped limiter and caches #20738

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.

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

Merged
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
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
12 changes: 9 additions & 3 deletions ydb/core/base/memory_controller_iface.h
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@ namespace NKikimr::NMemory {
enum class EMemoryConsumerKind {
SharedCache,
MemTable,
ScanMemoryLimiter,
CompMemoryLimiter,
BlobCache,
DataAccessorCache
};

struct IMemoryConsumer : public TThrRefBase {
Expand All @@ -17,7 +21,7 @@ enum EEvMemory {
EvConsumerRegister = EventSpaceBegin(TKikimrEvents::ES_MEMORY),
EvConsumerRegistered,
EvConsumerLimit,

EvMemTableRegister,
EvMemTableRegistered,
EvMemTableCompact,
Expand Down Expand Up @@ -47,10 +51,12 @@ struct TEvConsumerRegistered : public TEventLocal<TEvConsumerRegistered, EvConsu

struct TEvConsumerLimit : public TEventLocal<TEvConsumerLimit, EvConsumerLimit> {
ui64 LimitBytes;
std::optional<ui64> HardLimitBytes;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LimitBytes и так работает как hard limit

не надо сюда ничего такого было добавлять, к тому же по использованию вижу что вы просто домнажаете hard limit на коэффицент чтобы получить soft

просто делайте это внутри вашего memory consumer'а


TEvConsumerLimit(ui64 limitBytes)
TEvConsumerLimit(ui64 limitBytes, std::optional<ui64> hardLimitBytes = std::nullopt)
: LimitBytes(limitBytes)
{}
, HardLimitBytes(std::move(hardLimitBytes)) {
}
};

struct TEvMemTableRegister : public TEventLocal<TEvMemTableRegister, EvMemTableRegister> {
Expand Down
119 changes: 67 additions & 52 deletions ydb/core/memory_controller/memory_controller.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,6 @@
#include <ydb/core/tx/columnshard/blob_cache.h>
#include <ydb/core/tx/columnshard/common/limits.h>
#include <ydb/core/tx/columnshard/data_accessor/cache_policy/policy.h>
#include <ydb/core/tx/limiter/grouped_memory/usage/events.h>
#include <ydb/core/tx/limiter/grouped_memory/usage/service.h>
#include <ydb/library/actors/core/actor_bootstrapped.h>
#include <ydb/library/actors/core/log.h>
#include <ydb/library/actors/core/process_stats.h>
Expand Down Expand Up @@ -86,6 +84,7 @@ struct TConsumerState {
ui64 MinBytes = 0;
ui64 MaxBytes = 0;
bool CanZeroLimit = false;
std::optional<ui64> ExactLimit;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

если нужен был точный лимит, то можно просто MinBytes=MaxBytes выставить и не усложнять весь пересчёт ещё одним полем сбоку

а для того чтобы метрика была одна (без Min/Max) добавить флаг аналогично CanZeroLimit


TConsumerState(const TMemoryConsumer& consumer)
: Kind(consumer.Kind)
Expand Down Expand Up @@ -266,21 +265,31 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {

ui64 consumersLimitBytes = 0;
for (const auto& consumer : consumers) {
ui64 limitBytes = consumer.GetLimit(coefficient);
if (resultingConsumersConsumption + otherConsumption + externalConsumption > softLimitBytes && consumer.CanZeroLimit) {
limitBytes = SafeDiff(limitBytes, resultingConsumersConsumption + otherConsumption + externalConsumption - softLimitBytes);
const bool isExactLimitConsumer = consumer.ExactLimit.has_value();
ui64 limitBytes;
if (isExactLimitConsumer) {
limitBytes = consumer.ExactLimit.value();
} else {
limitBytes = consumer.GetLimit(coefficient);
if (resultingConsumersConsumption + otherConsumption + externalConsumption > softLimitBytes && consumer.CanZeroLimit) {
limitBytes = SafeDiff(limitBytes, resultingConsumersConsumption + otherConsumption + externalConsumption - softLimitBytes);
}
}
consumersLimitBytes += limitBytes;

LOG_INFO_S(ctx, NKikimrServices::MEMORY_CONTROLLER, "Consumer " << consumer.Kind << " state:"
<< " Consumption: " << HumanReadableBytes(consumer.Consumption) << " Limit: " << HumanReadableBytes(limitBytes)
<< " Min: " << HumanReadableBytes(consumer.MinBytes) << " Max: " << HumanReadableBytes(consumer.MaxBytes));
auto& counters = GetConsumerCounters(consumer.Kind);
auto& counters = GetConsumerCounters(consumer.Kind, !isExactLimitConsumer);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

я бы ожидал что аргумент будет просто называться так же как флаг, без лишней инверсии

counters.Consumption->Set(consumer.Consumption);
counters.Reservation->Set(SafeDiff(limitBytes, consumer.Consumption));
counters.LimitBytes->Set(limitBytes);
counters.LimitMinBytes->Set(consumer.MinBytes);
counters.LimitMaxBytes->Set(consumer.MaxBytes);
if (counters.LimitMinBytes) {
counters.LimitMinBytes->Set(consumer.MinBytes);
}
if (counters.LimitMaxBytes) {
counters.LimitMaxBytes->Set(consumer.MaxBytes);
}
SetMemoryStats(consumer, memoryStats, limitBytes);

ApplyLimit(consumer, limitBytes);
Expand All @@ -289,8 +298,6 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
Counters->GetCounter("Stats/ConsumersLimit")->Set(consumersLimitBytes);

ProcessResourceBrokerConfig(ctx, memoryStats, hardLimitBytes, activitiesLimitBytes);
ProcessGroupedMemoryLimiterConfig(ctx, memoryStats, hardLimitBytes);
ProcessCacheConfig(ctx, memoryStats, hardLimitBytes);

Send(NNodeWhiteboard::MakeNodeWhiteboardServiceId(SelfId().NodeId()), memoryStatsUpdate);

Expand Down Expand Up @@ -361,10 +368,15 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
case EMemoryConsumerKind::MemTable:
ApplyMemTableLimit(limitBytes);
break;
default:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

вернуть как было с default соответственно

case EMemoryConsumerKind::SharedCache:
case EMemoryConsumerKind::BlobCache:
case EMemoryConsumerKind::DataAccessorCache:
Send(consumer.ActorId, new TEvConsumerLimit(limitBytes));
break;

case EMemoryConsumerKind::ScanMemoryLimiter:
case EMemoryConsumerKind::CompMemoryLimiter:
Send(consumer.ActorId, new TEvConsumerLimit(limitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient, limitBytes));
break;
}
}

Expand All @@ -377,43 +389,6 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
}
}

void ProcessGroupedMemoryLimiterConfig(const TActorContext& /*ctx*/, NKikimrMemory::TMemoryStats& /*memoryStats*/, ui64 hardLimitBytes) {
ui64 columnTablesScanLimitBytes = GetColumnTablesReadExecutionLimitBytes(Config, hardLimitBytes);
ui64 columnTablesCompactionLimitBytes = GetColumnTablesCompactionLimitBytes(Config, hardLimitBytes) *
NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterCompactionLimitCoefficient;

ApplyGroupedMemoryLimiterConfig(columnTablesScanLimitBytes, columnTablesCompactionLimitBytes);
}

void ApplyGroupedMemoryLimiterConfig(const ui64 scanHardLimitBytes, const ui64 compactionHardLimitBytes) {
namespace NGroupedMemoryManager = ::NKikimr::NOlap::NGroupedMemoryManager;
using UpdateMemoryLimitsEv = NGroupedMemoryManager::NEvents::TEvExternal::TEvUpdateMemoryLimits;

Send(NGroupedMemoryManager::TScanMemoryLimiterOperator::MakeServiceId(SelfId().NodeId()),
new UpdateMemoryLimitsEv(scanHardLimitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient,
scanHardLimitBytes));

Send(NGroupedMemoryManager::TCompMemoryLimiterOperator::MakeServiceId(SelfId().NodeId()),
new UpdateMemoryLimitsEv(compactionHardLimitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient,
compactionHardLimitBytes));
}

void ProcessCacheConfig(const TActorContext& /*ctx*/, NKikimrMemory::TMemoryStats& /*memoryStats*/, ui64 hardLimitBytes) {
ui64 columnTablesBlobCacheLimitBytes = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes);

ApplyCacheConfig(columnTablesBlobCacheLimitBytes);
}

void ApplyCacheConfig(const ui64 cacheLimitBytes) {
using TUpdateMaxBlobCacheDataSizeEv = NBlobCache::TEvBlobCache::TEvUpdateMaxCacheDataSize;
using TGeneralCache = NKikimr::NOlap::NDataAccessorControl::TGeneralCache;
using TGlobalLimits = NKikimr::NOlap::TGlobalLimits;

Send(NKikimr::NBlobCache::MakeBlobCacheServiceId(), new TUpdateMaxBlobCacheDataSizeEv(cacheLimitBytes * TGlobalLimits::BlobCacheCoefficient));

TGeneralCache::UpdateMaxCacheSize(cacheLimitBytes * TGlobalLimits::DataAccessorCoefficient);
}

void ProcessResourceBrokerConfig(const TActorContext& ctx, NKikimrMemory::TMemoryStats& memoryStats, ui64 hardLimitBytes,
ui64 activitiesLimitBytes) {
ui64 queryExecutionConsumption = TAlignedPagePool::GetGlobalPagePoolSize();
Expand Down Expand Up @@ -470,18 +445,21 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
queue->MutableLimit()->SetMemory(limitBytes);
}

TConsumerCounters& GetConsumerCounters(EMemoryConsumerKind consumer) {
TConsumerCounters& GetConsumerCounters(EMemoryConsumerKind consumer, const bool minMaxRequired) {
auto it = ConsumerCounters.FindPtr(consumer);
if (it) {
return *it;
}

TCounterPtr limitMinBytes = minMaxRequired ? Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMin") : nullptr;
TCounterPtr limitMaxBytes = minMaxRequired ? Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMax") : nullptr;

return ConsumerCounters.emplace(consumer, TConsumerCounters{
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Consumption"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Reservation"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Limit"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMin"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMax"),
limitMinBytes,
limitMaxBytes,
}).first->second;
}

Expand All @@ -497,6 +475,26 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
stats.SetSharedCacheLimit(limitBytes);
break;
}
case EMemoryConsumerKind::ScanMemoryLimiter: {
stats.SetScanMemoryLimiterConsumption(consumer.Consumption);
stats.SetScanMemoryLimiterLimit(limitBytes);
break;
}
case EMemoryConsumerKind::CompMemoryLimiter: {
stats.SetCompMemoryLimiterConsumption(consumer.Consumption);
stats.SetCompMemoryLimiterLimit(limitBytes);
break;
}
case EMemoryConsumerKind::BlobCache: {
stats.SetBlobCacheConsumption(consumer.Consumption);
stats.SetBlobCacheLimit(limitBytes);
break;
}
case EMemoryConsumerKind::DataAccessorCache: {
stats.SetDataAccessorCacheConsumption(consumer.Consumption);
stats.SetDataAccessorCacheLimit(limitBytes);
break;
}
default:
Y_ABORT("Unhandled consumer");
}
Expand All @@ -517,6 +515,23 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
result.CanZeroLimit = true;
break;
}
case EMemoryConsumerKind::ScanMemoryLimiter: {
result.ExactLimit = GetColumnTablesReadExecutionLimitBytes(Config, hardLimitBytes);
break;
}
case EMemoryConsumerKind::CompMemoryLimiter: {
result.ExactLimit = GetColumnTablesCompactionLimitBytes(Config, hardLimitBytes) *
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

я бы ожидал что коэфицент NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterCompactionLimitCoefficient будет учтён внутри метода GetColumnTablesCompactionLimitBytes

думаю это можно легко сделать реализовав метод GetColumnTablesCompactionLimitBytes в memory_controller/memory_controller_config.h, внутри перевызвав GetColumnTablesCompactionLimitBytes который будет например в internal namespace и домножит его результат

Copy link
Member

@kunga kunga Jul 16, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

а ух, у нас вышло что часть лимитов названа так же как настройки, а часть нет, сложно(

тогда лучше сделать рядом с друг дружкой методы в memory_controller_config.h типа GetColumnTablesCompaction???LimitBytes, GetColumnTablesCompactionGeneralQueueLimitBytes и т.д

и коэфиценты перенести к ним же

так как по сути коэфиценты это то как конфиг преобразовывается в наши консьюмеры

NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterCompactionLimitCoefficient;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

using NKikimr::NOlap::TGlobalLimits в файле можно сделать

break;
}
case EMemoryConsumerKind::BlobCache: {
result.ExactLimit = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes) * NKikimr::NOlap::TGlobalLimits::BlobCacheCoefficient;
break;
}
case EMemoryConsumerKind::DataAccessorCache: {
result.ExactLimit = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes) * NKikimr::NOlap::TGlobalLimits::DataAccessorCoefficient;
break;
}
default:
Y_ABORT("Unhandled consumer");
}
Expand Down
11 changes: 11 additions & 0 deletions ydb/core/protos/memory_stats.proto
Original file line number Diff line number Diff line change
Expand Up @@ -27,4 +27,15 @@ message TMemoryStats {

optional uint64 QueryExecutionConsumption = 18;
optional uint64 QueryExecutionLimit = 19;

optional uint64 ScanMemoryLimiterConsumption = 20;
optional uint64 ScanMemoryLimiterLimit = 21;
optional uint64 CompMemoryLimiterConsumption = 22;
optional uint64 CompMemoryLimiterLimit = 23;
optional uint64 BlobCacheConsumption = 24;
optional uint64 BlobCacheLimit = 25;
optional uint64 DataAccessorCacheConsumption = 26;
optional uint64 DataAccessorCacheLimit = 27;


}
5 changes: 0 additions & 5 deletions ydb/core/testlib/test_client.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1207,11 +1207,6 @@ namespace Tests {
const auto aid = Runtime->Register(actor, nodeIdx, appData.UserPoolId, TMailboxType::Revolving, 0);
Runtime->RegisterService(NConveyorComposite::TServiceOperator::MakeServiceId(Runtime->GetNodeId(nodeIdx)), aid, nodeIdx);
}
{
auto* actor = NOlap::NDataAccessorControl::TGeneralCache::CreateService(NGeneralCache::NPublic::TConfig::BuildDefault(), new ::NMonitoring::TDynamicCounters());
const auto aid = Runtime->Register(actor, nodeIdx, appData.UserPoolId, TMailboxType::Revolving, 0);
Runtime->RegisterService(NOlap::NDataAccessorControl::TGeneralCache::MakeServiceId(Runtime->GetNodeId(nodeIdx)), aid, nodeIdx);
}
Runtime->Register(CreateLabelsMaintainer({}), nodeIdx, appData.SystemPoolId, TMailboxType::Revolving, 0);

auto sysViewService = NSysView::CreateSysViewServiceForTests();
Expand Down
30 changes: 27 additions & 3 deletions ydb/core/tx/columnshard/blob_cache.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

#include <ydb/core/base/appdata.h>
#include <ydb/core/base/blobstorage.h>
#include <ydb/core/base/memory_controller_iface.h>
#include <ydb/core/base/tablet_pipe.h>

#include <ydb/library/actors/core/actor.h>
Expand Down Expand Up @@ -134,6 +135,8 @@ class TBlobCache: public TActorBootstrapped<TBlobCache> {
const TCounterPtr ReadRequests;
const TCounterPtr ReadsInQueue;

TIntrusivePtr<NMemory::IMemoryConsumer> MemoryConsumer;

public:
static constexpr auto ActorActivityType() {
return NKikimrServices::TActivity::BLOB_CACHE_ACTOR;
Expand Down Expand Up @@ -178,6 +181,8 @@ class TBlobCache: public TActorBootstrapped<TBlobCache> {
LOG_S_NOTICE("MaxCacheDataSize: " << (i64)MaxCacheDataSize
<< " InFlightDataSize: " << (i64)InFlightDataSize);

Send(NMemory::MakeMemoryControllerId(), new NMemory::TEvConsumerRegister(NMemory::EMemoryConsumerKind::BlobCache));

Become(&TBlobCache::StateFunc);
ScheduleWakeup();
}
Expand All @@ -191,10 +196,11 @@ class TBlobCache: public TActorBootstrapped<TBlobCache> {
HFunc(TEvBlobCache::TEvReadBlobRangeBatch, Handle);
HFunc(TEvBlobCache::TEvCacheBlobRange, Handle);
HFunc(TEvBlobCache::TEvForgetBlob, Handle);
HFunc(TEvBlobCache::TEvUpdateMaxCacheDataSize, Handle);
HFunc(TEvBlobStorage::TEvGetResult, Handle);
HFunc(TEvTabletPipe::TEvClientConnected, Handle);
HFunc(TEvTabletPipe::TEvClientDestroyed, Handle);
HFunc(NMemory::TEvConsumerRegistered, Handle);
HFunc(NMemory::TEvConsumerLimit, Handle);
default:
LOG_S_WARN("Unhandled event type: " << ev->GetTypeRewrite()
<< " event: " << ev->ToString());
Expand Down Expand Up @@ -332,10 +338,16 @@ class TBlobCache: public TActorBootstrapped<TBlobCache> {
}

CachedRanges.erase(begin, end);

UpdateConsumption();
}

void Handle(TEvBlobCache::TEvUpdateMaxCacheDataSize::TPtr& ev, const TActorContext&) {
const i64 newMaxCacheDataSize = ev->Get()->MaxCacheDataSize;
void Handle(NMemory::TEvConsumerRegistered::TPtr& ev, const TActorContext&) {
MemoryConsumer = std::move(ev->Get()->Consumer);
}

void Handle(NMemory::TEvConsumerLimit::TPtr& ev, const TActorContext&) {
const i64 newMaxCacheDataSize = ev->Get()->LimitBytes;
if (newMaxCacheDataSize == (i64)MaxCacheDataSize) {
return;
}
Expand All @@ -345,6 +357,14 @@ class TBlobCache: public TActorBootstrapped<TBlobCache> {
MaxCacheDataSize = newMaxCacheDataSize;
}

void UpdateConsumption() {
if (!MemoryConsumer) {
return;
}

MemoryConsumer->SetConsumption(CacheDataSize);
}

void SendBatchReadRequestToDS(const std::vector<TBlobRange>& blobRanges, const ui64 cookie,
ui32 dsGroup, TReadItem::EReadVariant readVariant, const TActorContext& ctx)
{
Expand Down Expand Up @@ -579,6 +599,8 @@ class TBlobCache: public TActorBootstrapped<TBlobCache> {
SizeBytes->Add(blobRange.Size);
SizeBlobs->Inc();
}

UpdateConsumption();
}

void Evict(const TActorContext&) {
Expand All @@ -603,6 +625,8 @@ class TBlobCache: public TActorBootstrapped<TBlobCache> {
SizeBytes->Set(CacheDataSize);
SizeBlobs->Set(Cache.Size());
}

UpdateConsumption();
}
};

Expand Down
8 changes: 0 additions & 8 deletions ydb/core/tx/columnshard/blob_cache.h
Original file line number Diff line number Diff line change
Expand Up @@ -109,14 +109,6 @@ struct TEvBlobCache {
: BlobId(blobId)
{}
};

struct TEvUpdateMaxCacheDataSize: public NActors::TEventLocal<TEvUpdateMaxCacheDataSize, EvUpdateMemoryLimit> {
i64 MaxCacheDataSize = 0;

explicit TEvUpdateMaxCacheDataSize(const i64 maxCacheDataSize)
: MaxCacheDataSize(maxCacheDataSize) {
}
};
};

inline
Expand Down
6 changes: 6 additions & 0 deletions ydb/core/tx/columnshard/data_accessor/cache_policy/policy.h
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
#pragma once

#include <ydb/core/base/memory_controller_iface.h>
#include <ydb/core/tx/columnshard/engines/portions/data_accessor.h>
#include <ydb/core/tx/columnshard/engines/portions/meta.h>
#include <ydb/core/tx/general_cache/source/abstract.h>
Expand Down Expand Up @@ -72,6 +74,10 @@ class TPortionsMetadataCachePolicy {

static std::shared_ptr<NKikimr::NGeneralCache::NSource::IObjectsProcessor<TPortionsMetadataCachePolicy>> BuildObjectsProcessor(
const NActors::TActorId& serviceActorId);

static NMemory::EMemoryConsumerKind GetConsumerKind() {
return NMemory::EMemoryConsumerKind::DataAccessorCache;
}
};

} // namespace NKikimr::NOlap::NGeneralCache
Expand Down
Loading
Loading