Fixed scheduler fails

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

@@ -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();