Skip to content

Commit f5564a7

Browse files
ssmikessmike
authored
Merge 5656b12 into ce9f20d 
2 parents ce9f20d + 5656b12 commit f5564a7

File tree

7 files changed

+141
-15
lines changed

7 files changed

+141
-15
lines changed

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -241,6 +241,9 @@ std::unique_ptr<TEvKqpNode::TEvStartKqpTasksRequest> TKqpPlanner::SerializeReque
241241
if (UserRequestContext->PoolConfig->QueryCpuLimitPercentPerNode >= 0) {
242242
request.SetQueryCpuShare(UserRequestContext->PoolConfig->QueryCpuLimitPercentPerNode / 100.0);
243243
}
244+
if (UserRequestContext->PoolConfig->ResourceWeight >= 0) {
245+
request.SetResourceWeight(UserRequestContext->PoolConfig->ResourceWeight);
246+
}
244247
}
245248

246249
return result;

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

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -203,11 +203,16 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
203203

204204
if (SchedulerOptions.Scheduler->Disabled(schedulerGroup)) {
205205
auto share = msg.GetPoolMaxCpuShare();
206-
if (share <= 0 && msg.HasQueryCpuShare()) {
206+
if (share <= 0 && (msg.HasQueryCpuShare() || msg.HasResourceWeight())) {
207207
share = 1.0;
208208
}
209+
std::optional<double> resourceWeight;
210+
if (msg.GetResourceWeight() >= 0) {
211+
resourceWeight = msg.GetResourceWeight();
212+
}
213+
209214
if (share > 0) {
210-
Scheduler->UpdateGroupShare(schedulerGroup, share, schedulerNow);
215+
Scheduler->UpdateGroupShare(schedulerGroup, share, schedulerNow, resourceWeight);
211216
Send(SchedulerActorId, new TEvSchedulerNewPool(msg.GetDatabase(), schedulerGroup));
212217
} else {
213218
schedulerGroup = "";

‎ydb/core/kqp/runtime/kqp_compute_scheduler.cpp

Lines changed: 127 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,7 @@ class TObservableUpdater : IObservable {
171171
}
172172

173173
template<typename T>
174-
TParameter<T>* FindOrAddParameter(TParameterKey key, double def);
174+
TParameter<T>* FindOrAddParameter(TParameterKey key, T def);
175175

176176
ui64 ValuesCount() {
177177
return Params.size();
@@ -223,29 +223,32 @@ class TParameter : public IObservableValue<T> {
223223
Updater_->ToUpdate(this);
224224
}
225225

226-
void SetValue(T val) {
226+
T SetValue(T val) {
227+
auto oldValue = Value_;
227228
Value_ = val;
228229
Updater_->ToUpdate(this);
230+
return oldValue;
229231
}
230232

231233
protected:
232234
double DoUpdateValue() override {
233235
return Value_;
234236
}
235237

236-
private:
237238
T Value_;
239+
240+
private:
238241
TObservableUpdater* Updater_;
239242
};
240243

241244
template<typename T>
242-
TParameter<T>* TObservableUpdater::FindOrAddParameter(TParameterKey key, double def) {
245+
TParameter<T>* TObservableUpdater::FindOrAddParameter(TParameterKey key, T def) {
243246
if (auto* ptr = FindValue<TParameter<T>>(key)) {
244247
return ptr;
245248
}
246249
auto value = MakeHolder<TParameter<T>>(this, def);
247250
auto* result = value.Get();
248-
AddValue<TParameter<double>>(key, std::move(value));
251+
AddValue<TParameter<T>>(key, std::move(value));
249252
return result;
250253
}
251254

@@ -345,6 +348,62 @@ TSchedulerEntityHandle& TSchedulerEntityHandle::operator = (TSchedulerEntityHand
345348

346349
TSchedulerEntityHandle::~TSchedulerEntityHandle() = default;
347350

351+
class TSumResourceWeightsHolder : public TParameter<double> {
352+
public:
353+
using TParameter<double>::TParameter;
354+
355+
TSumResourceWeightsHolder(TObservableUpdater* engine)
356+
: TParameter(engine, 0)
357+
{
358+
}
359+
360+
void HandleUpdate(double delta) {
361+
SetValue(Value_ + delta);
362+
}
363+
364+
};
365+
366+
class TResourceWeightsUpdater {
367+
public:
368+
TResourceWeightsUpdater(TParameter<double>* param, double initial)
369+
: Param(param)
370+
, Value_(initial)
371+
{
372+
param->SetValue(Value_);
373+
}
374+
375+
void Track(TSumResourceWeightsHolder* holder) {
376+
if (!Holder) {
377+
Holder = holder;
378+
Holder->HandleUpdate(GetValue());
379+
}
380+
}
381+
382+
void Untrack() {
383+
if (Holder) {
384+
Holder->HandleUpdate(-GetValue());
385+
Holder = nullptr;
386+
}
387+
}
388+
389+
double GetValue() {
390+
return Value_;
391+
}
392+
393+
void SetValue(double val) {
394+
if (Holder) {
395+
Holder->HandleUpdate(val - Value_);
396+
}
397+
Value_ = val;
398+
Param->SetValue(Value_);
399+
}
400+
401+
private:
402+
TSumResourceWeightsHolder* Holder = nullptr;
403+
TParameter<double>* Param;
404+
double Value_;
405+
};
406+
348407
class TSchedulerEntity {
349408
public:
350409
TSchedulerEntity() {}
@@ -371,6 +430,7 @@ class TSchedulerEntity {
371430
std::atomic<i64> DelayedCount = 0;
372431

373432
THolder<IObservableValue<double>> Share;
433+
THolder<TResourceWeightsUpdater> ResourceWeightUpdater;
374434

375435
::NMonitoring::TDynamicCounters::TCounterPtr Vtime;
376436
::NMonitoring::TDynamicCounters::TCounterPtr EntitiesWeight;
@@ -495,10 +555,17 @@ struct TComputeScheduler::TImpl {
495555

496556
TObservableUpdater WeightsUpdater;
497557
TParameter<double> SumCores{&WeightsUpdater, 1};
558+
TSumResourceWeightsHolder SumResourceWeights{&WeightsUpdater};
498559

499560
enum : ui32 {
500-
Share = 1,
561+
TotalShare = 1,
562+
501563
PerQueryShare = 2,
564+
565+
ResourceWeight = 3,
566+
ResourceWeightEnabled = 4,
567+
568+
CompositeShare = 5,
502569
};
503570

504571
TIntrusivePtr<TKqpCounters> Counters;
@@ -565,6 +632,9 @@ void TComputeScheduler::AddToGroup(TMonotonic now, ui64 id, TSchedulerEntityHand
565632
auto group = Impl->Records[id].get();
566633
(*handle).Groups.push_back(group);
567634
group->MutableStats.Next()->EntitiesWeight += (*handle).Weight;
635+
if ((*handle).Weight > 0 && group->ResourceWeightUpdater) {
636+
group->ResourceWeightUpdater->Track(&Impl->SumResourceWeights);
637+
}
568638
Impl->AdvanceTime(now, group);
569639
}
570640

@@ -584,6 +654,7 @@ void TComputeScheduler::TImpl::AdvanceTime(TMonotonic now, TSchedulerEntity::TGr
584654
if (Counters) {
585655
record->InitCounters(Counters);
586656
}
657+
WeightsUpdater.UpdateAll();
587658
record->MutableStats.Next()->Capacity = record->Share->GetValue();
588659
auto& v = record->MutableStats;
589660
{
@@ -621,7 +692,6 @@ void TComputeScheduler::TImpl::AdvanceTime(TMonotonic now, TSchedulerEntity::TGr
621692
}
622693

623694
void TComputeScheduler::AdvanceTime(TMonotonic now) {
624-
Impl->WeightsUpdater.UpdateAll();
625695
for (size_t i = 0; i < Impl->Records.size(); ++i) {
626696
Impl->AdvanceTime(now, Impl->Records[i].get());
627697
}
@@ -636,6 +706,9 @@ void TComputeScheduler::Deregister(TSchedulerEntityHandle& self, TMonotonic now)
636706
for (auto group : (*self).Groups) {
637707
auto* next = group->MutableStats.Next();
638708
next->EntitiesWeight -= (*self).Weight;
709+
if (next->EntitiesWeight <= 0) {
710+
group->ResourceWeightUpdater->Untrack();
711+
}
639712
Impl->AdvanceTime(now, group);
640713
}
641714
}
@@ -701,14 +774,52 @@ void TComputeScheduler::Disable(TString group, TMonotonic now) {
701774
}
702775
}
703776

704-
void TComputeScheduler::UpdateGroupShare(TString group, double share, TMonotonic now) {
777+
class TCompositeGroupShare : public IObservableValue<double> {
778+
protected:
779+
double DoUpdateValue() override {
780+
if (ResourceWeightEnabled->GetValue()) {
781+
return Min(TotalLimit->GetValue(), ResourceWeight->GetValue() / SumResourceWeights->GetValue());
782+
} else {
783+
return TotalLimit->GetValue();
784+
}
785+
}
786+
787+
public:
788+
TCompositeGroupShare(IObservableValue<double>* resourceWeight, IObservableValue<bool>* resourceWeightEnabled, IObservableValue<double>* sumResourceWeights, IObservableValue<double>* totalLimit)
789+
: ResourceWeight(resourceWeight)
790+
, ResourceWeightEnabled(resourceWeightEnabled)
791+
, SumResourceWeights(sumResourceWeights)
792+
, TotalLimit(totalLimit)
793+
{
794+
Update();
795+
}
796+
797+
private:
798+
IObservableValue<double>* ResourceWeight;
799+
IObservableValue<bool>* ResourceWeightEnabled;
800+
IObservableValue<double>* SumResourceWeights;
801+
IObservableValue<double>* TotalLimit;
802+
};
803+
804+
void TComputeScheduler::UpdateGroupShare(TString group, double share, TMonotonic now, std::optional<double> resourceWeight) {
705805
auto ptr = Impl->GroupId.FindPtr(group);
706806

707-
auto* shareValue = Impl->WeightsUpdater.FindOrAddParameter<double>({group, TImpl::Share}, share);
807+
auto* shareValue = Impl->WeightsUpdater.FindOrAddParameter<double>({group, TImpl::TotalShare}, share);
708808
shareValue->SetValue(share);
809+
810+
TParameter<bool>* weightEnabled = Impl->WeightsUpdater.FindOrAddParameter<bool>({group, TImpl::ResourceWeightEnabled}, resourceWeight.has_value());
811+
weightEnabled->SetValue(resourceWeight.has_value());
812+
813+
TParameter<double>* resourceWeightValue = Impl->WeightsUpdater.FindOrAddParameter<double>({group, TImpl::ResourceWeight}, resourceWeight.value_or(0));
814+
709815
if (!ptr) {
710-
auto cap = MakeHolder<TShare>(&Impl->SumCores, shareValue);
816+
auto compositeWeight = MakeHolder<TCompositeGroupShare>(resourceWeightValue, weightEnabled, &Impl->SumResourceWeights, shareValue);
817+
auto resourceWeightsUpdater = MakeHolder<TResourceWeightsUpdater>(resourceWeightValue, resourceWeight.value_or(0));
818+
auto cap = MakeHolder<TShare>(&Impl->SumCores, compositeWeight.Get());
819+
Impl->WeightsUpdater.AddValue({group, TImpl::CompositeShare}, std::move(compositeWeight));
711820
Impl->CreateGroup(std::move(cap), now, group);
821+
822+
Impl->Records.back()->ResourceWeightUpdater = std::move(resourceWeightsUpdater);
712823
} else {
713824
auto& record = Impl->Records[*ptr];
714825
record->MutableStats.Next()->Disabled = false;
@@ -816,16 +927,20 @@ class TSchedulerActor : public TActorBootstrapped<TSchedulerActor> {
816927
if (ev->Get()->Config.has_value()) {
817928
auto totalShare = ev->Get()->Config->TotalCpuLimitPercentPerNode / 100.0;
818929
auto queryShare = ev->Get()->Config->QueryCpuLimitPercentPerNode / 100.0;
930+
std::optional<double> resourceWeight;
931+
if (ev->Get()->Config->ResourceWeight >= 0) {
932+
resourceWeight = ev->Get()->Config->ResourceWeight;
933+
}
819934

820-
if (totalShare <= 0 && queryShare > 0) {
935+
if (totalShare <= 0 && (queryShare > 0 || resourceWeight)) {
821936
totalShare = 1;
822937
}
823938

824939
if (queryShare <= 0) {
825940
queryShare = 1;
826941
}
827942

828-
Opts.Scheduler->UpdateGroupShare(ev->Get()->PoolId, totalShare, TlsActivationContext->Monotonic());
943+
Opts.Scheduler->UpdateGroupShare(ev->Get()->PoolId, totalShare, TlsActivationContext->Monotonic(), resourceWeight);
829944
Opts.Scheduler->UpdatePerQueryShare(ev->Get()->PoolId, queryShare, TlsActivationContext->Monotonic());
830945
} else {
831946
Opts.Scheduler->Disable(ev->Get()->PoolId, TlsActivationContext->Monotonic());

‎ydb/core/kqp/runtime/kqp_compute_scheduler.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ class TComputeScheduler {
6464

6565
void SetCapacity(ui64 cores);
6666

67-
void UpdateGroupShare(TString name, double share, TMonotonic now);
67+
void UpdateGroupShare(TString name, double share, TMonotonic now, std::optional<double> resourceWeight);
6868
void UpdatePerQueryShare(TString name, double share, TMonotonic now);
6969

7070
ui64 MakePerQueryGroup(TMonotonic now, double share, TString baseGroup);

‎ydb/core/protos/kqp.proto

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -558,6 +558,7 @@ message TEvStartKqpTasksRequest {
558558

559559
optional double PoolMaxCpuShare = 12;
560560
optional double QueryCpuShare = 16;
561+
optional double ResourceWeight = 17;
561562
}
562563

563564
message TEvStartKqpTasksResponse {

‎ydb/core/resource_pools/resource_pool_settings.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ std::unordered_map<TString, TPoolSettings::TProperty> TPoolSettings::GetProperti
5959
{"database_load_cpu_threshold", &DatabaseLoadCpuThreshold},
6060
{"total_cpu_limit_percent_per_node", &TotalCpuLimitPercentPerNode},
6161
{"query_cpu_limit_percent_per_node", &QueryCpuLimitPercentPerNode},
62+
{"resource_weight", &ResourceWeight}
6263
};
6364
if (!restricted) {
6465
properties.insert({"query_cancel_after_seconds", &QueryCancelAfter});

‎ydb/core/resource_pools/resource_pool_settings.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ struct TPoolSettings : public TSettingsBase {
4646
TPercent DatabaseLoadCpuThreshold = -1; // -1 = disabled
4747
TPercent TotalCpuLimitPercentPerNode = -1; // -1 = disabled
4848
TPercent QueryCpuLimitPercentPerNode = -1; // -1 = disabled;
49+
double ResourceWeight = -1;
4950
};
5051

5152
} // namespace NKikimr::NResourcePool

0 commit comments

Comments
 (0)