Skip to content

Commit 19ace32

Browse files
SammyVimesSammyVimes
authored
Inflight limit for ReadRows to prevent dynnodes OOM (#16019)
1 parent 38f45f8 commit 19ace32

File tree

9 files changed

+121
-31
lines changed

9 files changed

+121
-31
lines changed

‎ydb/core/cms/console/immediate_controls_configurator.cpp

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,8 @@ class TImmediateControlsConfigurator : public TActorBootstrapped<TImmediateContr
5151
TIntrusivePtr<TControlBoard> board);
5252
void ApplyConfig(const ::google::protobuf::Message &cfg,
5353
const TString &prefix,
54-
TIntrusivePtr<TControlBoard> board);
54+
TIntrusivePtr<TControlBoard> board,
55+
bool allowDynamicFields = false);
5556
TString MakePrefix(const TString &prefix,
5657
const TString &name);
5758

@@ -114,6 +115,11 @@ void TImmediateControlsConfigurator::CreateControls(TIntrusivePtr<TControlBoard>
114115
{
115116
for (int i = 0; i < desc->field_count(); ++i) {
116117
auto *fieldDesc = desc->field(i);
118+
auto name = MakePrefix(prefix, fieldDesc->name());
119+
120+
if (fieldDesc->is_map()) {
121+
continue;
122+
}
117123

118124
Y_ABORT_UNLESS(!fieldDesc->is_repeated(),
119125
"Repeated fields are not allowed in Immediate Controls Config");
@@ -171,17 +177,50 @@ void TImmediateControlsConfigurator::ApplyConfig(const NKikimrConfig::TImmediate
171177

172178
void TImmediateControlsConfigurator::ApplyConfig(const ::google::protobuf::Message &cfg,
173179
const TString &prefix,
174-
TIntrusivePtr<TControlBoard> board)
180+
TIntrusivePtr<TControlBoard> board,
181+
bool allowDynamicFields)
175182
{
176183
auto *desc = cfg.GetDescriptor();
177184
auto *reflection = cfg.GetReflection();
178185
for (int i = 0; i < desc->field_count(); ++i) {
179-
auto *fieldDesc = desc->field(i);
186+
const auto *fieldDesc = desc->field(i);
180187
auto fieldType = fieldDesc->type();
181188
auto name = MakePrefix(prefix, fieldDesc->name());
189+
190+
if (fieldDesc->is_map()) {
191+
auto *mapDesc = fieldDesc->message_type();
192+
auto *mapKey = mapDesc->map_key();
193+
auto *mapValue = mapDesc->map_value();
194+
195+
auto keyType = mapKey->type();
196+
auto valueType = mapValue->type();
197+
198+
Y_ABORT_UNLESS(keyType == google::protobuf::FieldDescriptor::TYPE_STRING,
199+
"Only string keys are allowed in Immediate Controls Config maps");
200+
201+
Y_ABORT_UNLESS(valueType == google::protobuf::FieldDescriptor::TYPE_MESSAGE,
202+
"Only message value are allowed in Immediate Controls Config maps");
203+
204+
auto entryCount = reflection->FieldSize(cfg, fieldDesc);
205+
for (int j = 0; j < entryCount; ++j) {
206+
const auto &entry = reflection->GetRepeatedMessage(cfg, fieldDesc, j);
207+
auto *entryReflection = entry.GetReflection();
208+
auto key = entryReflection->GetString(entry, mapKey);
209+
auto entryName = MakePrefix(name, key);
210+
ApplyConfig(entryReflection->GetMessage(entry, mapValue), entryName, board, true);
211+
}
212+
continue;
213+
}
214+
182215
if (fieldType == google::protobuf::FieldDescriptor::TYPE_UINT64
183216
|| fieldType == google::protobuf::FieldDescriptor::TYPE_INT64) {
184-
Y_ABORT_UNLESS(Controls.contains(name));
217+
if (!Controls.contains(name)) {
218+
if (!allowDynamicFields) {
219+
Y_ABORT("Missing control for field %s", name.c_str());
220+
}
221+
AddControl(board, fieldDesc, prefix, true);
222+
}
223+
185224
if (reflection->HasField(cfg, fieldDesc)) {
186225
TAtomicBase prev;
187226
if (fieldType == google::protobuf::FieldDescriptor::TYPE_UINT64)

‎ydb/core/cms/console/immediate_controls_configurator_ut.cpp

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -258,6 +258,32 @@ Y_UNIT_TEST_SUITE(TImmediateControlsConfiguratorTests)
258258
MakeAddAction(ITEM_CONTROLS_EXCEED_MAX));
259259
CompareControls(runtime, ITEM_CONTROLS_MAX.GetConfig().GetImmediateControlsConfig());
260260
}
261+
262+
Y_UNIT_TEST(TestDynamicMap)
263+
{
264+
TTenantTestRuntime runtime(DefaultConsoleTestConfig());
265+
InitImmediateControlsConfigurator(runtime);
266+
WaitForUpdate(runtime); // initial update
267+
268+
NKikimrConsole::TConfigItem dynamicMapValue;
269+
{
270+
auto &cfg = *dynamicMapValue.MutableConfig()->MutableImmediateControlsConfig();
271+
auto *grpcControls = cfg.MutableGRpcControls();
272+
273+
auto *requestConfigs = grpcControls->MutableRequestConfigs();
274+
auto &r = (*requestConfigs)["FooBar"];
275+
r.SetMaxInFlight(10);
276+
}
277+
278+
ConfigureAndWaitUpdate(runtime, MakeAddAction(dynamicMapValue));
279+
280+
auto icb = runtime.GetAppData().Icb;
281+
282+
TControlWrapper wrapper;
283+
284+
icb->RegisterSharedControl(wrapper, "GRpcControls.RequestConfigs.FooBar.MaxInFlight");
285+
UNIT_ASSERT_VALUES_EQUAL((ui64)(i64)wrapper, 10);
286+
}
261287
}
262288

263289
} // namespace NKikimr

‎ydb/core/driver_lib/run/run.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -759,8 +759,8 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
759759
}
760760

761761
if (hasTableService) {
762-
server.AddService(new NGRpcService::TGRpcYdbTableService(ActorSystem.Get(), Counters, grpcRequestProxies,
763-
hasTableService.IsRlAllowed(), grpcConfig.GetHandlersPerCompletionQueue()));
762+
server.AddService(new NGRpcService::TGRpcYdbTableService(ActorSystem.Get(), Counters, AppData->InFlightLimiterRegistry,
763+
grpcRequestProxies, hasTableService.IsRlAllowed(), grpcConfig.GetHandlersPerCompletionQueue()));
764764
}
765765

766766
if (hasClickhouseInternal) {

‎ydb/core/grpc_services/grpc_helper.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ namespace NGRpcService {
66
//using namespace NActors;
77

88
NYdbGrpc::IGRpcRequestLimiterPtr TCreateLimiterCB::operator()(const char* serviceName, const char* requestName, i64 limit) const {
9-
TString fullName = TString(serviceName) + "_" + requestName;
9+
TString fullName = "GRpcControls.RequestConfigs." + TString(serviceName) + "_" + requestName;
1010
return LimiterRegistry->RegisterRequestType(fullName, limit);
1111
}
1212

@@ -34,7 +34,7 @@ NYdbGrpc::IGRpcRequestLimiterPtr TInFlightLimiterRegistry::RegisterRequestType(T
3434
TGuard<TMutex> g(Lock);
3535
if (!PerTypeLimiters.count(name)) {
3636
TControlWrapper control(limit, 0, 1000000);
37-
Icb->RegisterSharedControl(control, name + "_MaxInFlight");
37+
Icb->RegisterSharedControl(control, name + ".MaxInFlight");
3838
PerTypeLimiters[name] = new TRequestInFlightLimiter(control);
3939
}
4040

‎ydb/core/protos/config.proto

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1573,6 +1573,18 @@ message TImmediateControlsConfig {
15731573
DefaultValue: 0 }];
15741574
}
15751575

1576+
message TGRpcControls {
1577+
message TRequestConfig {
1578+
optional uint64 MaxInFlight = 1 [(ControlOptions) = {
1579+
Description: "Max in flight requests",
1580+
MinValue: 0,
1581+
MaxValue: 1000000,
1582+
DefaultValue: 0
1583+
}];
1584+
}
1585+
map<string, TRequestConfig> RequestConfigs = 1;
1586+
}
1587+
15761588
optional TDataShardControls DataShardControls = 1;
15771589
optional TTxLimitControls TxLimitControls = 2;
15781590
optional TCoordinatorControls CoordinatorControls = 3;
@@ -1585,6 +1597,7 @@ message TImmediateControlsConfig {
15851597
optional TPDiskControls PDiskControls = 10;
15861598
optional TBlobStorageControllerControls BlobStorageControllerControls = 11;
15871599
optional TTableServiceControls TableServiceControls = 12;
1600+
optional TGRpcControls GRpcControls = 14;
15881601
};
15891602

15901603
message TMeteringConfig {

‎ydb/core/testlib/test_client.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -699,7 +699,7 @@ namespace Tests {
699699
GRpcServer->AddService(new NGRpcService::TGRpcYdbExportService(system, counters, grpcRequestProxies[0], true));
700700
GRpcServer->AddService(new NGRpcService::TGRpcYdbImportService(system, counters, grpcRequestProxies[0], true));
701701
GRpcServer->AddService(new NGRpcService::TGRpcYdbSchemeService(system, counters, grpcRequestProxies[0], true));
702-
GRpcServer->AddService(new NGRpcService::TGRpcYdbTableService(system, counters, grpcRequestProxies, true, 1));
702+
GRpcServer->AddService(new NGRpcService::TGRpcYdbTableService(system, counters, appData.InFlightLimiterRegistry, grpcRequestProxies, true, 1));
703703
GRpcServer->AddService(new NGRpcService::TGRpcYdbScriptingService(system, counters, grpcRequestProxies[0], true));
704704
GRpcServer->AddService(new NGRpcService::TGRpcOperationService(system, counters, grpcRequestProxies[0], true));
705705
GRpcServer->AddService(new NGRpcService::V1::TGRpcPersQueueService(system, counters, NMsgBusProxy::CreatePersQueueMetaCacheV2Id(), grpcRequestProxies[0], true));

‎ydb/library/grpc/server/grpc_request.h

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -601,8 +601,9 @@ class TGRpcRequest: public TGRpcRequestImpl<TIn, TOut, TService, TInProtoPrinter
601601
typename TBase::TStreamRequestCallback requestCallback,
602602
const char* name,
603603
TLoggerPtr logger,
604-
ICounterBlockPtr counters)
605-
: TBase{server, service, cq, std::move(cb), std::move(requestCallback), name, std::move(logger), std::move(counters), nullptr}
604+
ICounterBlockPtr counters,
605+
IGRpcRequestLimiterPtr limiter = nullptr)
606+
: TBase{server, service, cq, std::move(cb), std::move(requestCallback), name, std::move(logger), std::move(counters), std::move(limiter)}
606607
{
607608
}
608609
};

‎ydb/services/ydb/ydb_table.cpp

Lines changed: 26 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -9,26 +9,31 @@ namespace NGRpcService {
99

1010
TGRpcYdbTableService::TGRpcYdbTableService(NActors::TActorSystem *system,
1111
TIntrusivePtr<::NMonitoring::TDynamicCounters> counters,
12+
TIntrusivePtr<TInFlightLimiterRegistry> inFlightLimiterRegistry,
1213
const NActors::TActorId& proxyId,
1314
bool rlAllowed,
1415
size_t handlersPerCompletionQueue)
1516
: TGrpcServiceBase(system, counters, proxyId, rlAllowed)
1617
, HandlersPerCompletionQueue(Max(size_t{1}, handlersPerCompletionQueue))
18+
, LimiterRegistry_(inFlightLimiterRegistry)
1719
{
1820
}
1921

2022
TGRpcYdbTableService::TGRpcYdbTableService(NActors::TActorSystem *system,
2123
TIntrusivePtr<::NMonitoring::TDynamicCounters> counters,
24+
TIntrusivePtr<TInFlightLimiterRegistry> inFlightLimiterRegistry,
2225
const TVector<NActors::TActorId>& proxies,
2326
bool rlAllowed,
2427
size_t handlersPerCompletionQueue)
2528
: TGrpcServiceBase(system, counters, proxies, rlAllowed)
2629
, HandlersPerCompletionQueue(Max(size_t{1}, handlersPerCompletionQueue))
30+
, LimiterRegistry_(inFlightLimiterRegistry)
2731
{
2832
}
2933

3034
void TGRpcYdbTableService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) {
3135
auto getCounterBlock = CreateCounterCb(Counters_, ActorSystem_);
36+
auto getLimiter = CreateLimiterCb(LimiterRegistry_);
3237

3338
size_t proxyCounter = 0;
3439

@@ -60,23 +65,24 @@ void TGRpcYdbTableService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) {
6065
} \
6166
}
6267

63-
#define ADD_STREAM_REQUEST_LIMIT(NAME, IN, OUT, CB, LIMIT_TYPE, REQUEST_TYPE) \
64-
for (size_t i = 0; i < HandlersPerCompletionQueue; ++i) { \
65-
for (auto* cq: CQS) { \
66-
MakeIntrusive<TGRpcRequest<Ydb::Table::IN, Ydb::Table::OUT, TGRpcYdbTableService>> \
67-
(this, &Service_, cq, \
68-
[this, proxyCounter](NYdbGrpc::IRequestContextBase *ctx) { \
69-
NGRpcService::ReportGrpcReqToMon(*ActorSystem_, ctx->GetPeer()); \
70-
ActorSystem_->Send(GRpcProxies_[proxyCounter % GRpcProxies_.size()], \
71-
new TGrpcRequestNoOperationCall<Ydb::Table::IN, Ydb::Table::OUT> \
72-
(ctx, &CB, TRequestAuxSettings { \
73-
.RlMode = RLSWITCH(TRateLimiterMode::LIMIT_TYPE), \
74-
.RequestType = NJaegerTracing::ERequestType::TABLE_##REQUEST_TYPE, \
75-
})); \
76-
}, &Ydb::Table::V1::TableService::AsyncService::Request ## NAME, \
77-
#NAME, logger, getCounterBlock("table", #NAME))->Run(); \
78-
++proxyCounter; \
79-
} \
68+
#define ADD_STREAM_REQUEST_LIMIT(NAME, IN, OUT, CB, LIMIT_TYPE, REQUEST_TYPE, USE_LIMITER) \
69+
for (size_t i = 0; i < HandlersPerCompletionQueue; ++i) { \
70+
for (auto* cq: CQS) { \
71+
MakeIntrusive<TGRpcRequest<Ydb::Table::IN, Ydb::Table::OUT, TGRpcYdbTableService>> \
72+
(this, &Service_, cq, \
73+
[this, proxyCounter](NYdbGrpc::IRequestContextBase *ctx) { \
74+
NGRpcService::ReportGrpcReqToMon(*ActorSystem_, ctx->GetPeer()); \
75+
ActorSystem_->Send(GRpcProxies_[proxyCounter % GRpcProxies_.size()], \
76+
new TGrpcRequestNoOperationCall<Ydb::Table::IN, Ydb::Table::OUT> \
77+
(ctx, &CB, TRequestAuxSettings { \
78+
.RlMode = RLSWITCH(TRateLimiterMode::LIMIT_TYPE), \
79+
.RequestType = NJaegerTracing::ERequestType::TABLE_##REQUEST_TYPE, \
80+
})); \
81+
}, &Ydb::Table::V1::TableService::AsyncService::Request ## NAME, \
82+
#NAME, logger, getCounterBlock("table", #NAME), \
83+
(USE_LIMITER ? getLimiter("TableService", #NAME, UNLIMITED_INFLIGHT) : nullptr))->Run(); \
84+
++proxyCounter; \
85+
} \
8086
}
8187

8288
ADD_REQUEST_LIMIT(CreateSession, DoCreateSessionRequest, Rps, CREATESESSION)
@@ -103,9 +109,9 @@ void TGRpcYdbTableService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) {
103109
ADD_REQUEST_LIMIT(ExecuteDataQuery, DoExecuteDataQueryRequest, Ru, EXECUTEDATAQUERY, Auditable)
104110
ADD_REQUEST_LIMIT(BulkUpsert, DoBulkUpsertRequest, Ru, BULKUPSERT, Auditable)
105111

106-
ADD_STREAM_REQUEST_LIMIT(StreamExecuteScanQuery, ExecuteScanQueryRequest, ExecuteScanQueryPartialResponse, DoExecuteScanQueryRequest, RuOnProgress, STREAMEXECUTESCANQUERY)
107-
ADD_STREAM_REQUEST_LIMIT(StreamReadTable, ReadTableRequest, ReadTableResponse, DoReadTableRequest, RuOnProgress, STREAMREADTABLE)
108-
ADD_STREAM_REQUEST_LIMIT(ReadRows, ReadRowsRequest, ReadRowsResponse, DoReadRowsRequest, Ru, READROWS)
112+
ADD_STREAM_REQUEST_LIMIT(StreamExecuteScanQuery, ExecuteScanQueryRequest, ExecuteScanQueryPartialResponse, DoExecuteScanQueryRequest, RuOnProgress, STREAMEXECUTESCANQUERY, false)
113+
ADD_STREAM_REQUEST_LIMIT(StreamReadTable, ReadTableRequest, ReadTableResponse, DoReadTableRequest, RuOnProgress, STREAMREADTABLE, false)
114+
ADD_STREAM_REQUEST_LIMIT(ReadRows, ReadRowsRequest, ReadRowsResponse, DoReadRowsRequest, Ru, READROWS, true)
109115

110116
#undef ADD_REQUEST_LIMIT
111117
#undef ADD_STREAM_REQUEST_LIMIT

‎ydb/services/ydb/ydb_table.h

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,26 +4,30 @@
44
#include <ydb/library/grpc/server/grpc_server.h>
55
#include <ydb/public/api/grpc/ydb_table_v1.grpc.pb.h>
66
#include <ydb/core/grpc_services/base/base_service.h>
7+
#include <ydb/core/grpc_services/grpc_helper.h>
78

89
namespace NKikimr {
910
namespace NGRpcService {
1011

1112
class TGRpcYdbTableService
1213
: public TGrpcServiceBase<Ydb::Table::V1::TableService>
1314
{
15+
constexpr static i64 UNLIMITED_INFLIGHT = 0;
1416
public:
1517
using TGrpcServiceBase<Ydb::Table::V1::TableService>::TGrpcServiceBase;
1618

1719
TGRpcYdbTableService(
1820
NActors::TActorSystem *system,
1921
TIntrusivePtr<::NMonitoring::TDynamicCounters> counters,
22+
TIntrusivePtr<TInFlightLimiterRegistry> inFlightLimiterRegistry,
2023
const NActors::TActorId& proxyId,
2124
bool rlAllowed,
2225
size_t handlersPerCompletionQueue = 1);
2326

2427
TGRpcYdbTableService(
2528
NActors::TActorSystem *system,
2629
TIntrusivePtr<::NMonitoring::TDynamicCounters> counters,
30+
TIntrusivePtr<TInFlightLimiterRegistry> inFlightLimiterRegistry,
2731
const TVector<NActors::TActorId>& proxies,
2832
bool rlAllowed,
2933
size_t handlersPerCompletionQueue);
@@ -33,6 +37,7 @@ class TGRpcYdbTableService
3337

3438
private:
3539
const size_t HandlersPerCompletionQueue;
40+
TIntrusivePtr<NGRpcService::TInFlightLimiterRegistry> LimiterRegistry_;
3641
};
3742

3843
} // namespace NGRpcService

0 commit comments

Comments
 (0)