merge to stable kqp cpu scheduler fix use after free (#20155)

Author: GrigoriyPA

ydb/core/kqp/node_service/kqp_node_service.cpp

@@ -210,7 +210,7 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
                 share = 1.0;
             }
             std::optional<double> resourceWeight;
-            if (msg.GetResourceWeight() >= 0) {
+            if (msg.HasResourceWeight() && msg.GetResourceWeight() >= 0) {
                 resourceWeight = msg.GetResourceWeight();
             }
 

ydb/core/kqp/runtime/kqp_compute_scheduler.cpp

@@ -188,14 +188,17 @@ class TObservableUpdater {
 
     void CollectValues() {
         std::vector<TParameterKey> toerase;
-        for (auto& [k, v] : Params) {
-            if (!v.Holder->HasDependents()) {
-                toerase.push_back(k);
+        do {
+            toerase.clear();
+            for (auto& [k, v] : Params) {
+                if (!v.Holder->HasDependents()) {
+                    toerase.push_back(k);
+                }
             }
-        }
-        for (auto& key : toerase) {
-            Params.erase(key);
-        }
+            for (auto& key : toerase) {
+                Params.erase(key);
+            }
+        } while (!toerase.empty());
     }
 
 private:
@@ -897,7 +900,8 @@ class TCompositeGroupShare : public IObservableValue<double> {
 protected:
     double DoUpdateValue() override {
         if (ResourceWeightEnabled->GetValue()) {
-            if (ResourceWeightLimit->Enabled()->GetValue()) {
+            auto limitEnabled = ResourceWeightLimit->Enabled();
+            if (limitEnabled->GetValue()) {
                 return Min(TotalLimit->GetValue(), ResourceWeightLimit->GetValue());
             } else {
                 return 0;

ydb/core/tx/conveyor/service/service.cpp

@@ -10,7 +10,7 @@ TWorkersPool::TWorkersPool(const TString& conveyorName, const NActors::TActorId&
     Workers.reserve(WorkersCount);
     for (ui32 i = 0; i < WorkersCount; ++i) {
         const auto usage = config.GetWorkerCPUUsage(i);
-        Workers.emplace_back(std::make_unique<TWorker>(conveyorName, usage, distributorId, i, Counters.SendFwdHistogram, Counters.SendFwdDuration));
+        Workers.emplace_back(usage, std::make_unique<TWorker>(conveyorName, usage, distributorId, i, Counters.SendFwdHistogram, Counters.SendFwdDuration));
         MaxWorkerThreads += usage;
     }
     AFL_VERIFY(WorkersCount)("name", conveyorName)("action", "conveyor_registered")("config", config.DebugString())("actor_id", distributorId)("count", WorkersCount);
@@ -46,6 +46,9 @@ void TWorkersPool::ReleaseWorker(const ui32 workerIdx) {
 
 void TWorkersPool::ChangeAmountCPULimit(const double delta) {
     AmountCPULimit += delta;
+    if (std::abs(AmountCPULimit) < Eps) {
+        AmountCPULimit = 0;
+    }
     AFL_VERIFY(AmountCPULimit >= 0);
     Counters.AmountCPULimit->Set(AmountCPULimit);
     Counters.ChangeCPULimitRate->Inc();
@@ -58,15 +61,15 @@ void TWorkersPool::ChangeAmountCPULimit(const double delta) {
     ActiveWorkersCount = 0;
     ActiveWorkerThreads = numberThreads;
     ActiveWorkersIdx.clear();
-    for (const auto& worker : Workers) {
+    for (auto& worker : Workers) {
         if (numberThreads <= 0) {
             break;
         }
         if (!worker.GetRunningTask()) {
             ActiveWorkersIdx.emplace_back(ActiveWorkersCount);
         }
-        worker.GetWorker()->UpdateCPUSoftLimit(std::min<double>(numberThreads, 1));
-        numberThreads -= worker.GetWorker()->GetCPUSoftLimit();
+        worker.ChangeCPUSoftLimit(std::min<double>(numberThreads, 1));
+        numberThreads -= worker.GetCPUSoftLimit();
         ++ActiveWorkersCount;
     }
     AFL_VERIFY(std::abs(numberThreads) < Eps);

ydb/core/tx/conveyor/service/service.h

@@ -188,14 +188,21 @@ class TProcess {
 class TWorkersPool {
     class TWorkerInfo {
         YDB_READONLY(bool, RunningTask, false);
-        YDB_READONLY(TWorker*, Worker, nullptr);
+        YDB_READONLY(double, CPUSoftLimit, 0.0);
         YDB_READONLY_DEF(TActorId, WorkerId);
     public:
-        explicit TWorkerInfo(std::unique_ptr<TWorker> worker)
-            : Worker(worker.get())
+        TWorkerInfo(const double cpuSoftLimit, std::unique_ptr<TWorker> worker)
+            : CPUSoftLimit(cpuSoftLimit)
             , WorkerId(TActivationContext::Register(worker.release())) {
         }
 
+        void ChangeCPUSoftLimit(const double newCPUSoftLimit) {
+            if (std::abs(newCPUSoftLimit - CPUSoftLimit) > Eps) {
+                TActivationContext::Send(WorkerId, std::make_unique<TEvInternal::TEvChangeCPUSoftLimit>(newCPUSoftLimit));
+                CPUSoftLimit = newCPUSoftLimit;
+            }
+        }
+
         void OnStartTask() {
             AFL_VERIFY(!RunningTask);
             RunningTask = true;

ydb/core/tx/conveyor/service/worker.cpp

@@ -62,7 +62,8 @@ void TWorker::HandleMain(TEvInternal::TEvNewTask::TPtr& ev) {
     ExecuteTask(ev->Get()->ExtractTasks());
 }
 
-void TWorker::UpdateCPUSoftLimit(const double cpuSoftLimit) {
+void TWorker::HandleMain(TEvInternal::TEvChangeCPUSoftLimit::TPtr& ev) {
+    const auto cpuSoftLimit = ev->Get()->GetCPUSoftLimit();
     AFL_VERIFY(0 < cpuSoftLimit);
     AFL_VERIFY(cpuSoftLimit <= CPUHardLimit);
     CPUSoftLimit = cpuSoftLimit;

ydb/core/tx/conveyor/service/worker.h

@@ -44,6 +44,7 @@ struct TEvInternal {
     enum EEv {
         EvNewTask = EventSpaceBegin(NActors::TEvents::ES_PRIVATE),
         EvTaskProcessedResult,
+        EvChangeCPUSoftLimit,
         EvEnd
     };
 
@@ -84,6 +85,15 @@ struct TEvInternal {
             AFL_VERIFY(Instants.size() == ProcessIds.size() + 1);
         }
     };
+
+    class TEvChangeCPUSoftLimit: public NActors::TEventLocal<TEvChangeCPUSoftLimit, EvChangeCPUSoftLimit> {
+    private:
+        YDB_READONLY(double, CPUSoftLimit, 0.0);
+    public:
+        explicit TEvChangeCPUSoftLimit(const double cpuSoftLimit)
+            : CPUSoftLimit(cpuSoftLimit) {
+        }
+    };
 };
 
 class TWorker: public NActors::TActorBootstrapped<TWorker> {
@@ -104,13 +114,15 @@ class TWorker: public NActors::TActorBootstrapped<TWorker> {
     void ExecuteTask(std::vector<TWorkerTask>&& workerTasks);
     void HandleMain(TEvInternal::TEvNewTask::TPtr& ev);
     void HandleMain(NActors::TEvents::TEvWakeup::TPtr& ev);
+    void HandleMain(TEvInternal::TEvChangeCPUSoftLimit::TPtr& ev);
     void OnWakeup();
 public:
 
     STATEFN(StateMain) {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvInternal::TEvNewTask, HandleMain);
             hFunc(NActors::TEvents::TEvWakeup, HandleMain);
+            hFunc(TEvInternal::TEvChangeCPUSoftLimit, HandleMain);
             default:
                 ALS_ERROR(NKikimrServices::TX_CONVEYOR) << "unexpected event for task executor: " << ev->GetTypeRewrite();
                 break;
@@ -132,8 +144,6 @@ class TWorker: public NActors::TActorBootstrapped<TWorker> {
         AFL_VERIFY(0 < CPUHardLimit);
         AFL_VERIFY(CPUHardLimit <= 1);
     }
-
-    void UpdateCPUSoftLimit(const double cpuSoftLimit);
 };
 
 }