Skip to content

Commit 597e16a

Browse files
VladilenVladilen
authored
Add stats for grouped limiter and caches (#20738)
1 parent 1122775 commit 597e16a

File tree

20 files changed

+233
-125
lines changed

20 files changed

+233
-125
lines changed

‎ydb/core/base/memory_controller_iface.h

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,10 @@ namespace NKikimr::NMemory {
77
enum class EMemoryConsumerKind {
88
SharedCache,
99
MemTable,
10+
ScanGroupedMemoryLimiter,
11+
CompGroupedMemoryLimiter,
12+
BlobCache,
13+
DataAccessorCache
1014
};
1115

1216
struct IMemoryConsumer : public TThrRefBase {
@@ -17,7 +21,7 @@ enum EEvMemory {
1721
EvConsumerRegister = EventSpaceBegin(TKikimrEvents::ES_MEMORY),
1822
EvConsumerRegistered,
1923
EvConsumerLimit,
20-
24+
2125
EvMemTableRegister,
2226
EvMemTableRegistered,
2327
EvMemTableCompact,
@@ -47,10 +51,12 @@ struct TEvConsumerRegistered : public TEventLocal<TEvConsumerRegistered, EvConsu
4751

4852
struct TEvConsumerLimit : public TEventLocal<TEvConsumerLimit, EvConsumerLimit> {
4953
ui64 LimitBytes;
54+
std::optional<ui64> HardLimitBytes;
5055

51-
TEvConsumerLimit(ui64 limitBytes)
56+
TEvConsumerLimit(ui64 limitBytes, std::optional<ui64> hardLimitBytes = std::nullopt)
5257
: LimitBytes(limitBytes)
53-
{}
58+
, HardLimitBytes(std::move(hardLimitBytes)) {
59+
}
5460
};
5561

5662
struct TEvMemTableRegister : public TEventLocal<TEvMemTableRegister, EvMemTableRegister> {

‎ydb/core/base/ut/memory_stats_ut.cpp

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
6969
aggregator.Add(stats1, "host1");
7070
aggregator.Add(stats2, "host2");
7171
aggregator.Add(stats3, "host3");
72-
72+
7373
TMemoryStats aggregated = aggregator.Aggregate();
7474

7575
Cerr << aggregated.ShortDebugString() << Endl;
@@ -96,7 +96,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
9696
aggregator.Add(stats1, "host1");
9797
aggregator.Add(stats2, "host2");
9898
aggregator.Add(stats3, "host3");
99-
99+
100100
TMemoryStats aggregated = aggregator.Aggregate();
101101

102102
Cerr << aggregated.ShortDebugString() << Endl;
@@ -119,7 +119,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
119119
aggregator.Add(stats1, "host");
120120
aggregator.Add(stats2, "host");
121121
aggregator.Add(stats3, "host");
122-
122+
123123
TMemoryStats aggregated = aggregator.Aggregate();
124124

125125
Cerr << aggregated.ShortDebugString() << Endl;
@@ -146,7 +146,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
146146
aggregator.Add(stats1, "host");
147147
aggregator.Add(stats2, "host");
148148
aggregator.Add(stats3, "host");
149-
149+
150150
TMemoryStats aggregated = aggregator.Aggregate();
151151

152152
Cerr << aggregated.ShortDebugString() << Endl;
@@ -169,14 +169,32 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
169169
aggregator.Add(stats1, "host1");
170170
aggregator.Add(stats2, "host1");
171171
aggregator.Add(stats3, "host2");
172-
172+
173173
TMemoryStats aggregated = aggregator.Aggregate();
174174

175175
Cerr << aggregated.ShortDebugString() << Endl;
176176

177177
UNIT_ASSERT_VALUES_EQUAL(aggregated.ShortDebugString(),
178178
"AnonRss: 36 CGroupLimit: 66 MemTotal: 65 MemAvailable: 85 AllocatedMemory: 156 AllocatorCachesMemory: 186 HardLimit: 145 SoftLimit: 165 TargetUtilization: 185 ExternalConsumption: 194 SharedCacheConsumption: 336 SharedCacheLimit: 366 MemTableConsumption: 396 MemTableLimit: 426 QueryExecutionConsumption: 456 QueryExecutionLimit: 486");
179179
}
180+
181+
Y_UNIT_TEST(ColumnShard_Single) {
182+
TMemoryStatsAggregator aggregator;
183+
184+
TMemoryStats stats;
185+
stats.SetColumnTablesReadExecutionConsumption(1);
186+
stats.SetColumnTablesReadExecutionLimit(2);
187+
stats.SetColumnTablesCompactionConsumption(3);
188+
stats.SetColumnTablesCompactionLimit(4);
189+
stats.SetColumnTablesCacheConsumption(5);
190+
stats.SetColumnTablesCacheLimit(6);
191+
192+
aggregator.Add(stats, "host");
193+
194+
TMemoryStats aggregated = aggregator.Aggregate();
195+
196+
UNIT_ASSERT_VALUES_EQUAL(aggregated.ShortDebugString(), stats.ShortDebugString());
197+
}
180198
}
181199

182200
}

‎ydb/core/memory_controller/memory_controller.cpp

Lines changed: 64 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,6 @@
1616
#include <ydb/core/tx/columnshard/blob_cache.h>
1717
#include <ydb/core/tx/columnshard/common/limits.h>
1818
#include <ydb/core/tx/columnshard/data_accessor/cache_policy/policy.h>
19-
#include <ydb/core/tx/limiter/grouped_memory/usage/events.h>
20-
#include <ydb/core/tx/limiter/grouped_memory/usage/service.h>
2119
#include <ydb/library/actors/core/actor_bootstrapped.h>
2220
#include <ydb/library/actors/core/log.h>
2321
#include <ydb/library/actors/core/process_stats.h>
@@ -86,6 +84,7 @@ struct TConsumerState {
8684
ui64 MinBytes = 0;
8785
ui64 MaxBytes = 0;
8886
bool CanZeroLimit = false;
87+
std::optional<ui64> ExactLimit;
8988

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

267266
ui64 consumersLimitBytes = 0;
268267
for (const auto& consumer : consumers) {
269-
ui64 limitBytes = consumer.GetLimit(coefficient);
270-
if (resultingConsumersConsumption + otherConsumption + externalConsumption > softLimitBytes && consumer.CanZeroLimit) {
271-
limitBytes = SafeDiff(limitBytes, resultingConsumersConsumption + otherConsumption + externalConsumption - softLimitBytes);
268+
const bool isExactLimitConsumer = consumer.ExactLimit.has_value();
269+
ui64 limitBytes;
270+
if (isExactLimitConsumer) {
271+
limitBytes = consumer.ExactLimit.value();
272+
} else {
273+
limitBytes = consumer.GetLimit(coefficient);
274+
if (resultingConsumersConsumption + otherConsumption + externalConsumption > softLimitBytes && consumer.CanZeroLimit) {
275+
limitBytes = SafeDiff(limitBytes, resultingConsumersConsumption + otherConsumption + externalConsumption - softLimitBytes);
276+
}
272277
}
273278
consumersLimitBytes += limitBytes;
274279

275280
LOG_INFO_S(ctx, NKikimrServices::MEMORY_CONTROLLER, "Consumer " << consumer.Kind << " state:"
276281
<< " Consumption: " << HumanReadableBytes(consumer.Consumption) << " Limit: " << HumanReadableBytes(limitBytes)
277282
<< " Min: " << HumanReadableBytes(consumer.MinBytes) << " Max: " << HumanReadableBytes(consumer.MaxBytes));
278-
auto& counters = GetConsumerCounters(consumer.Kind);
283+
auto& counters = GetConsumerCounters(consumer.Kind, !isExactLimitConsumer);
279284
counters.Consumption->Set(consumer.Consumption);
280285
counters.Reservation->Set(SafeDiff(limitBytes, consumer.Consumption));
281286
counters.LimitBytes->Set(limitBytes);
282-
counters.LimitMinBytes->Set(consumer.MinBytes);
283-
counters.LimitMaxBytes->Set(consumer.MaxBytes);
287+
if (counters.LimitMinBytes) {
288+
counters.LimitMinBytes->Set(consumer.MinBytes);
289+
}
290+
if (counters.LimitMaxBytes) {
291+
counters.LimitMaxBytes->Set(consumer.MaxBytes);
292+
}
284293
SetMemoryStats(consumer, memoryStats, limitBytes);
285294

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

291300
ProcessResourceBrokerConfig(ctx, memoryStats, hardLimitBytes, activitiesLimitBytes);
292-
ProcessGroupedMemoryLimiterConfig(ctx, memoryStats, hardLimitBytes);
293-
ProcessCacheConfig(ctx, memoryStats, hardLimitBytes);
294301

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

@@ -361,10 +368,15 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
361368
case EMemoryConsumerKind::MemTable:
362369
ApplyMemTableLimit(limitBytes);
363370
break;
364-
default:
371+
case EMemoryConsumerKind::SharedCache:
372+
case EMemoryConsumerKind::BlobCache:
373+
case EMemoryConsumerKind::DataAccessorCache:
365374
Send(consumer.ActorId, new TEvConsumerLimit(limitBytes));
366375
break;
367-
376+
case EMemoryConsumerKind::ScanGroupedMemoryLimiter:
377+
case EMemoryConsumerKind::CompGroupedMemoryLimiter:
378+
Send(consumer.ActorId, new TEvConsumerLimit(limitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient, limitBytes));
379+
break;
368380
}
369381
}
370382

@@ -377,43 +389,6 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
377389
}
378390
}
379391

380-
void ProcessGroupedMemoryLimiterConfig(const TActorContext& /*ctx*/, NKikimrMemory::TMemoryStats& /*memoryStats*/, ui64 hardLimitBytes) {
381-
ui64 columnTablesScanLimitBytes = GetColumnTablesReadExecutionLimitBytes(Config, hardLimitBytes);
382-
ui64 columnTablesCompactionLimitBytes = GetColumnTablesCompactionLimitBytes(Config, hardLimitBytes) *
383-
NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterCompactionLimitCoefficient;
384-
385-
ApplyGroupedMemoryLimiterConfig(columnTablesScanLimitBytes, columnTablesCompactionLimitBytes);
386-
}
387-
388-
void ApplyGroupedMemoryLimiterConfig(const ui64 scanHardLimitBytes, const ui64 compactionHardLimitBytes) {
389-
namespace NGroupedMemoryManager = ::NKikimr::NOlap::NGroupedMemoryManager;
390-
using UpdateMemoryLimitsEv = NGroupedMemoryManager::NEvents::TEvExternal::TEvUpdateMemoryLimits;
391-
392-
Send(NGroupedMemoryManager::TScanMemoryLimiterOperator::MakeServiceId(SelfId().NodeId()),
393-
new UpdateMemoryLimitsEv(scanHardLimitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient,
394-
scanHardLimitBytes));
395-
396-
Send(NGroupedMemoryManager::TCompMemoryLimiterOperator::MakeServiceId(SelfId().NodeId()),
397-
new UpdateMemoryLimitsEv(compactionHardLimitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient,
398-
compactionHardLimitBytes));
399-
}
400-
401-
void ProcessCacheConfig(const TActorContext& /*ctx*/, NKikimrMemory::TMemoryStats& /*memoryStats*/, ui64 hardLimitBytes) {
402-
ui64 columnTablesBlobCacheLimitBytes = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes);
403-
404-
ApplyCacheConfig(columnTablesBlobCacheLimitBytes);
405-
}
406-
407-
void ApplyCacheConfig(const ui64 cacheLimitBytes) {
408-
using TUpdateMaxBlobCacheDataSizeEv = NBlobCache::TEvBlobCache::TEvUpdateMaxCacheDataSize;
409-
using TGeneralCache = NKikimr::NOlap::NDataAccessorControl::TGeneralCache;
410-
using TGlobalLimits = NKikimr::NOlap::TGlobalLimits;
411-
412-
Send(NKikimr::NBlobCache::MakeBlobCacheServiceId(), new TUpdateMaxBlobCacheDataSizeEv(cacheLimitBytes * TGlobalLimits::BlobCacheCoefficient));
413-
414-
TGeneralCache::UpdateMaxCacheSize(cacheLimitBytes * TGlobalLimits::DataAccessorCoefficient);
415-
}
416-
417392
void ProcessResourceBrokerConfig(const TActorContext& ctx, NKikimrMemory::TMemoryStats& memoryStats, ui64 hardLimitBytes,
418393
ui64 activitiesLimitBytes) {
419394
ui64 queryExecutionConsumption = TAlignedPagePool::GetGlobalPagePoolSize();
@@ -470,18 +445,21 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
470445
queue->MutableLimit()->SetMemory(limitBytes);
471446
}
472447

473-
TConsumerCounters& GetConsumerCounters(EMemoryConsumerKind consumer) {
448+
TConsumerCounters& GetConsumerCounters(EMemoryConsumerKind consumer, const bool minMaxRequired) {
474449
auto it = ConsumerCounters.FindPtr(consumer);
475450
if (it) {
476451
return *it;
477452
}
478453

454+
TCounterPtr limitMinBytes = minMaxRequired ? Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMin") : nullptr;
455+
TCounterPtr limitMaxBytes = minMaxRequired ? Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMax") : nullptr;
456+
479457
return ConsumerCounters.emplace(consumer, TConsumerCounters{
480458
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Consumption"),
481459
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Reservation"),
482460
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Limit"),
483-
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMin"),
484-
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMax"),
461+
limitMinBytes,
462+
limitMaxBytes,
485463
}).first->second;
486464
}
487465

@@ -497,6 +475,23 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
497475
stats.SetSharedCacheLimit(limitBytes);
498476
break;
499477
}
478+
case EMemoryConsumerKind::ScanGroupedMemoryLimiter: {
479+
stats.SetColumnTablesReadExecutionConsumption(consumer.Consumption);
480+
stats.SetColumnTablesReadExecutionLimit(limitBytes);
481+
break;
482+
}
483+
case EMemoryConsumerKind::CompGroupedMemoryLimiter: {
484+
stats.SetColumnTablesCompactionConsumption(consumer.Consumption);
485+
stats.SetColumnTablesCompactionLimit(limitBytes);
486+
break;
487+
}
488+
case EMemoryConsumerKind::DataAccessorCache:
489+
case EMemoryConsumerKind::BlobCache: {
490+
stats.SetColumnTablesCacheConsumption(
491+
(stats.HasColumnTablesCacheConsumption() ? stats.GetColumnTablesCacheConsumption() : 0) + consumer.Consumption);
492+
stats.SetColumnTablesCacheLimit((stats.HasColumnTablesCacheLimit() ? stats.GetColumnTablesCacheLimit() : 0) + limitBytes);
493+
break;
494+
}
500495
default:
501496
Y_ABORT("Unhandled consumer");
502497
}
@@ -517,6 +512,23 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
517512
result.CanZeroLimit = true;
518513
break;
519514
}
515+
case EMemoryConsumerKind::ScanGroupedMemoryLimiter: {
516+
result.ExactLimit = GetColumnTablesReadExecutionLimitBytes(Config, hardLimitBytes);
517+
break;
518+
}
519+
case EMemoryConsumerKind::CompGroupedMemoryLimiter: {
520+
result.ExactLimit = GetColumnTablesCompactionLimitBytes(Config, hardLimitBytes) *
521+
NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterCompactionLimitCoefficient;
522+
break;
523+
}
524+
case EMemoryConsumerKind::BlobCache: {
525+
result.ExactLimit = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes) * NKikimr::NOlap::TGlobalLimits::BlobCacheCoefficient;
526+
break;
527+
}
528+
case EMemoryConsumerKind::DataAccessorCache: {
529+
result.ExactLimit = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes) * NKikimr::NOlap::TGlobalLimits::DataAccessorCoefficient;
530+
break;
531+
}
520532
default:
521533
Y_ABORT("Unhandled consumer");
522534
}

‎ydb/core/protos/memory_stats.proto

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,4 +27,12 @@ message TMemoryStats {
2727

2828
optional uint64 QueryExecutionConsumption = 18;
2929
optional uint64 QueryExecutionLimit = 19;
30+
31+
optional uint64 ColumnTablesReadExecutionConsumption = 20;
32+
optional uint64 ColumnTablesReadExecutionLimit = 21;
33+
optional uint64 ColumnTablesCompactionConsumption = 22;
34+
optional uint64 ColumnTablesCompactionLimit = 23;
35+
optional uint64 ColumnTablesCacheConsumption = 24;
36+
optional uint64 ColumnTablesCacheLimit = 25;
37+
3038
}

‎ydb/core/testlib/test_client.cpp

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1207,11 +1207,6 @@ namespace Tests {
12071207
const auto aid = Runtime->Register(actor, nodeIdx, appData.UserPoolId, TMailboxType::Revolving, 0);
12081208
Runtime->RegisterService(NConveyorComposite::TServiceOperator::MakeServiceId(Runtime->GetNodeId(nodeIdx)), aid, nodeIdx);
12091209
}
1210-
{
1211-
auto* actor = NOlap::NDataAccessorControl::TGeneralCache::CreateService(NGeneralCache::NPublic::TConfig::BuildDefault(), new ::NMonitoring::TDynamicCounters());
1212-
const auto aid = Runtime->Register(actor, nodeIdx, appData.UserPoolId, TMailboxType::Revolving, 0);
1213-
Runtime->RegisterService(NOlap::NDataAccessorControl::TGeneralCache::MakeServiceId(Runtime->GetNodeId(nodeIdx)), aid, nodeIdx);
1214-
}
12151210
Runtime->Register(CreateLabelsMaintainer({}), nodeIdx, appData.SystemPoolId, TMailboxType::Revolving, 0);
12161211

12171212
auto sysViewService = NSysView::CreateSysViewServiceForTests();

0 commit comments

Comments
 (0)