Skip to content

Commit 1500574

Browse files
ssmikessmike
authored
Use cpu limits from resource pool scheme objects (#8370)
1 parent bfadc55 commit 1500574

File tree

8 files changed

+227
-212
lines changed

8 files changed

+227
-212
lines changed

‎ydb/core/kqp/executer_actor/kqp_planner.cpp

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,10 @@ void BuildInitialTaskResources(const TKqpTasksGraph& graph, ui64 taskId, TTaskRe
5353
ret.HeavyProgram = opts.GetHasMapJoin();
5454
}
5555

56+
bool LimitCPU(TIntrusivePtr<TUserRequestContext> ctx) {
57+
return ctx->PoolId && ctx->PoolConfig.has_value() && ctx->PoolConfig->TotalCpuLimitPercentPerNode > 0;
58+
}
59+
5660
}
5761

5862
bool TKqpPlanner::UseMockEmptyPlanner = false;
@@ -101,6 +105,10 @@ TKqpPlanner::TKqpPlanner(TKqpPlanner::TArgs&& args)
101105
LOG_E("Database not set, use " << Database);
102106
}
103107
}
108+
109+
if (LimitCPU(UserRequestContext)) {
110+
AllowSinglePartitionOpt = false;
111+
}
104112
}
105113

106114
// ResourcesSnapshot, ResourceEstimations
@@ -226,6 +234,7 @@ std::unique_ptr<TEvKqpNode::TEvStartKqpTasksRequest> TKqpPlanner::SerializeReque
226234
request.SetDatabase(Database);
227235
if (UserRequestContext->PoolConfig.has_value()) {
228236
request.SetMemoryPoolPercent(UserRequestContext->PoolConfig->QueryMemoryLimitPercentPerNode);
237+
request.SetMaxCpuShare(UserRequestContext->PoolConfig->TotalCpuLimitPercentPerNode / 100.0);
229238
}
230239

231240
return result;

‎ydb/core/kqp/node_service/kqp_node_service.cpp

Lines changed: 37 additions & 79 deletions
Original file line numberDiff line numberDiff line change
@@ -81,12 +81,13 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
8181
if (config.HasIteratorReadQuotaSettings()) {
8282
SetIteratorReadsQuotaSettings(config.GetIteratorReadQuotaSettings());
8383
}
84-
if (config.HasPoolsConfiguration()) {
85-
SetPriorities(config.GetPoolsConfiguration());
86-
}
87-
Scheduler.ReportCounters(counters);
88-
AdvanceTimeInterval = TDuration::MicroSeconds(config.GetComputeSchedulerSettings().GetAdvanceTimeIntervalUsec());
89-
Scheduler.SetForgetInterval(TDuration::MicroSeconds(config.GetComputeSchedulerSettings().GetForgetOverflowTimeoutUsec()));
84+
85+
SchedulerOptions = {
86+
.AdvanceTimeInterval = TDuration::MicroSeconds(config.GetComputeSchedulerSettings().GetAdvanceTimeIntervalUsec()),
87+
.ForgetOverflowTimeout = TDuration::MicroSeconds(config.GetComputeSchedulerSettings().GetForgetOverflowTimeoutUsec()),
88+
.ActivePoolPollingTimeout = TDuration::Seconds(config.GetComputeSchedulerSettings().GetActivePoolPollingSec()),
89+
.Counters = counters,
90+
};
9091
}
9192

9293
void Bootstrap() {
@@ -105,15 +106,13 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
105106
TlsActivationContext->ExecutorThread.ActorSystem, SelfId());
106107
}
107108

108-
Schedule(TDuration::Seconds(1), new TEvents::TEvWakeup(WakeCleaunupTag));
109-
Schedule(AdvanceTimeInterval, new TEvents::TEvWakeup(WakeAdvanceTimeTag));
109+
Schedule(TDuration::Seconds(1), new TEvents::TEvWakeup());
110110
Become(&TKqpNodeService::WorkState);
111-
}
112111

113-
enum {
114-
WakeCleaunupTag,
115-
WakeAdvanceTimeTag
116-
};
112+
Scheduler = std::make_shared<TComputeScheduler>();
113+
SchedulerOptions.Scheduler = Scheduler;
114+
SchedulerActorId = RegisterWithSameMailbox(CreateSchedulerActor(SchedulerOptions));
115+
}
117116

118117
private:
119118
STATEFN(WorkState) {
@@ -128,8 +127,6 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
128127
hFunc(TEvents::TEvUndelivered, HandleWork);
129128
hFunc(TEvents::TEvPoison, HandleWork);
130129
hFunc(NMon::TEvHttpInfo, HandleWork);
131-
// sheduling
132-
hFunc(TEvSchedulerDeregister, HandleWork);
133130
default: {
134131
Y_ABORT("Unexpected event 0x%x for TKqpResourceManagerService", ev->GetTypeRewrite());
135132
}
@@ -138,12 +135,6 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
138135

139136
static constexpr double SecToUsec = 1e6;
140137

141-
void HandleWork(TEvSchedulerDeregister::TPtr& ev) {
142-
if (ev->Get()->SchedulerEntity) {
143-
Scheduler.Deregister(*ev->Get()->SchedulerEntity, TMonotonic::Now());
144-
}
145-
}
146-
147138
void HandleWork(TEvKqpNode::TEvStartKqpTasksRequest::TPtr& ev) {
148139
NWilson::TSpan sendTasksSpan(TWilsonKqp::KqpNodeSendTasks, NWilson::TTraceId(ev->TraceId), "KqpNode.SendTasks", NWilson::EFlags::AUTO_END);
149140

@@ -198,7 +189,7 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
198189
const TString& serializedGUCSettings = ev->Get()->Record.HasSerializedGUCSettings() ?
199190
ev->Get()->Record.GetSerializedGUCSettings() : "";
200191

201-
auto schedulerNow = TMonotonic::Now();
192+
auto schedulerNow = TlsActivationContext->Monotonic();
202193

203194
// start compute actors
204195
TMaybe<NYql::NDqProto::TRlPath> rlPath = Nothing();
@@ -215,20 +206,28 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
215206
for (auto& dqTask: *msg.MutableTasks()) {
216207
TString group = msg.GetSchedulerGroup();
217208

218-
TComputeActorSchedulingOptions schedulingOptions {
209+
TComputeActorSchedulingOptions schedulingTaskOptions {
219210
.Now = schedulerNow,
220-
.NodeService = SelfId(),
221-
.Scheduler = &Scheduler,
211+
.SchedulerActorId = SchedulerActorId,
212+
.Scheduler = Scheduler.get(),
222213
.Group = group,
223214
.Weight = 1,
224215
.NoThrottle = false,
225216
.Counters = Counters
226217
};
227218

228-
if (Scheduler.Disabled(schedulingOptions.Group)) {
229-
schedulingOptions.NoThrottle = true;
230-
} else {
231-
schedulingOptions.Handle = Scheduler.Enroll(schedulingOptions.Group, schedulingOptions.Weight, schedulingOptions.Now);
219+
if (SchedulerOptions.Scheduler->Disabled(group)) {
220+
auto share = msg.GetMaxCpuShare();
221+
if (share > 0) {
222+
Scheduler->UpdateMaxShare(group, share, schedulerNow);
223+
Send(SchedulerActorId, new TEvSchedulerNewPool(msg.GetDatabase(), group, share));
224+
} else {
225+
schedulingTaskOptions.NoThrottle = true;
226+
}
227+
}
228+
229+
if (!schedulingTaskOptions.NoThrottle) {
230+
schedulingTaskOptions.Handle = SchedulerOptions.Scheduler->Enroll(schedulingTaskOptions.Group, schedulingTaskOptions.Weight, schedulingTaskOptions.Now);
232231
}
233232

234233
auto result = CaFactory_->CreateKqpComputeActor({
@@ -250,7 +249,7 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
250249
.RlPath = rlPath,
251250
.ComputesByStages = &computesByStage,
252251
.State = State_,
253-
.SchedulingOptions = std::move(schedulingOptions),
252+
.SchedulingOptions = std::move(schedulingTaskOptions),
254253
});
255254

256255
if (const auto* rmResult = std::get_if<NRm::TKqpRMAllocateResult>(&result)) {
@@ -343,17 +342,11 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
343342
}
344343

345344
void HandleWork(TEvents::TEvWakeup::TPtr& ev) {
346-
if (ev->Get()->Tag == WakeAdvanceTimeTag) {
347-
Scheduler.AdvanceTime(TMonotonic::Now());
348-
Schedule(AdvanceTimeInterval, new TEvents::TEvWakeup(WakeAdvanceTimeTag));
349-
}
350-
if (ev->Get()->Tag == WakeCleaunupTag) {
351-
Schedule(TDuration::Seconds(1), ev->Release().Release());
352-
for (auto& bucket : State_->Buckets) {
353-
auto expiredRequests = bucket.ClearExpiredRequests();
354-
for (auto& cxt : expiredRequests) {
355-
TerminateTx(cxt.TxId, "reached execution deadline", NYql::NDqProto::StatusIds::TIMEOUT);
356-
}
345+
Schedule(TDuration::Seconds(1), ev->Release().Release());
346+
for (auto& bucket : State_->Buckets) {
347+
auto expiredRequests = bucket.ClearExpiredRequests();
348+
for (auto& cxt : expiredRequests) {
349+
TerminateTx(cxt.TxId, "reached execution deadline", NYql::NDqProto::StatusIds::TIMEOUT);
357350
}
358351
}
359352
}
@@ -395,13 +388,6 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
395388
SetIteratorReadsQuotaSettings(event.GetConfig().GetTableServiceConfig().GetIteratorReadQuotaSettings());
396389
}
397390

398-
if (event.GetConfig().GetTableServiceConfig().HasPoolsConfiguration()) {
399-
SetPriorities(event.GetConfig().GetTableServiceConfig().GetPoolsConfiguration());
400-
}
401-
402-
AdvanceTimeInterval = TDuration::MicroSeconds(event.GetConfig().GetTableServiceConfig().GetComputeSchedulerSettings().GetAdvanceTimeIntervalUsec());
403-
Scheduler.SetForgetInterval(TDuration::MicroSeconds(event.GetConfig().GetTableServiceConfig().GetComputeSchedulerSettings().GetForgetOverflowTimeoutUsec()));
404-
405391
auto responseEv = MakeHolder<NConsole::TEvConsole::TEvConfigNotificationResponse>(event);
406392
Send(ev->Sender, responseEv.Release(), IEventHandle::FlagTrackDelivery, ev->Cookie);
407393
}
@@ -410,35 +396,6 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
410396
SetDefaultIteratorQuotaSettings(settings.GetMaxRows(), settings.GetMaxBytes());
411397
}
412398

413-
void SetPriorities(const NKikimrConfig::TTableServiceConfig::TComputePoolConfiguration& conf) {
414-
std::function<TComputeScheduler::TDistributionRule(const NKikimrConfig::TTableServiceConfig::TComputePoolConfiguration&)> convert
415-
= [&](const NKikimrConfig::TTableServiceConfig::TComputePoolConfiguration& conf)
416-
{
417-
if (conf.HasName()) {
418-
return TComputeScheduler::TDistributionRule{.Share = conf.GetMaxCpuShare(), .Name = conf.GetName()};
419-
} else if (conf.HasSubPoolsConfiguration()) {
420-
auto res = TComputeScheduler::TDistributionRule{.Share = conf.GetMaxCpuShare()};
421-
for (auto& subConf : conf.GetSubPoolsConfiguration().GetSubPools()) {
422-
res.SubRules.push_back(convert(subConf));
423-
}
424-
return res;
425-
} else {
426-
Y_ENSURE(false, "unknown case");
427-
}
428-
};
429-
SetPriorities(convert(conf));
430-
}
431-
432-
void SetPriorities(TComputeScheduler::TDistributionRule rule) {
433-
NActors::TExecutorPoolStats poolStats;
434-
TVector<NActors::TExecutorThreadStats> threadsStats;
435-
TlsActivationContext->ActorSystem()->GetPoolStats(SelfId().PoolID(), poolStats, threadsStats);
436-
Y_ENSURE(poolStats.MaxThreadCount > 0);
437-
Counters->SchedulerCapacity->Set(poolStats.MaxThreadCount);
438-
439-
Scheduler.SetPriorities(rule, poolStats.MaxThreadCount, TMonotonic::Now());
440-
}
441-
442399
void SetIteratorReadsRetrySettings(const NKikimrConfig::TTableServiceConfig::TIteratorReadsRetrySettings& settings) {
443400
auto ptr = MakeIntrusive<NKikimr::NKqp::TIteratorReadBackoffSettings>();
444401
ptr->StartRetryDelay = TDuration::MilliSeconds(settings.GetStartDelayMs());
@@ -525,8 +482,9 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
525482
NYql::NDq::IDqAsyncIoFactory::TPtr AsyncIoFactory;
526483
const std::optional<TKqpFederatedQuerySetup> FederatedQuerySetup;
527484

528-
TComputeScheduler Scheduler;
529-
TDuration AdvanceTimeInterval;
485+
std::shared_ptr<TComputeScheduler> Scheduler;
486+
TSchedulerActorOptions SchedulerOptions;
487+
TActorId SchedulerActorId;
530488

531489
//state sharded by TxId
532490
std::shared_ptr<TNodeServiceState> State_;

0 commit comments

Comments
 (0)