Ydb stable 23-1-1923.1.19

x-stable-origin-commit: c5d5a396e89d0a72e0267a55e93d8404d4fb54fe

1 files changed, 38 insertions, 27 deletions

diff --git a/library/cpp/actors/core/harmonizer.cpp b/library/cpp/actors/core/harmonizer.cpp
index f318d8909c5..e2fd0c5f24a 100644
--- a/library/cpp/actors/core/harmonizer.cpp
+++ b/library/cpp/actors/core/harmonizer.cpp
@@ -121,6 +121,7 @@ struct TPoolInfo {
     TAtomic IncreasingThreadsByNeedyState = 0;
     TAtomic DecreasingThreadsByStarvedState = 0;
     TAtomic DecreasingThreadsByHoggishState = 0;
+    TAtomic PotentialMaxThreadCount = 0;
 
     bool IsBeingStopped(i16 threadIdx);
     double GetBooked(i16 threadIdx);
@@ -169,9 +170,10 @@ double TPoolInfo::GetlastSecondPoolConsumed(i16 threadIdx) {
 void TPoolInfo::PullStats(ui64 ts) {
     for (i16 threadIdx = 0; threadIdx < MaxThreadCount; ++threadIdx) {
         TThreadInfo &threadInfo = ThreadInfo[threadIdx];
-        threadInfo.Consumed.Register(ts, Pool->GetThreadConsumedUs(threadIdx));
+        TCpuConsumption cpuConsumption = Pool->GetThreadCpuConsumption(threadIdx);
+        threadInfo.Consumed.Register(ts, cpuConsumption.ConsumedUs);
         LWPROBE(SavedValues, Pool->PoolId, Pool->GetName(), "consumed", UNROLL_HISTORY(threadInfo.Consumed.History));
-        threadInfo.Booked.Register(ts, Pool->GetThreadBookedUs(threadIdx));
+        threadInfo.Booked.Register(ts, cpuConsumption.BookedUs);
         LWPROBE(SavedValues, Pool->PoolId, Pool->GetName(), "booked", UNROLL_HISTORY(threadInfo.Booked.History));
     }
 }
@@ -236,7 +238,7 @@ void THarmonizer::PullStats(ui64 ts) {
 }
 
 Y_FORCE_INLINE bool IsStarved(double consumed, double booked) {
-    return Max(consumed, booked) > 0.1 && consumed < booked * 0.7;
+    return consumed < booked * 0.7;
 }
 
 Y_FORCE_INLINE bool IsHoggish(double booked, ui16 currentThreadCount) {
@@ -293,35 +295,43 @@ void THarmonizer::HarmonizeImpl(ui64 ts) {
         LWPROBE(HarmonizeCheckPool, poolIdx, pool.Pool->GetName(), poolBooked, poolConsumed, lastSecondPoolBooked, lastSecondPoolConsumed, pool.GetThreadCount(), pool.MaxThreadCount, isStarved, isNeedy, isHoggish);
     }
     double budget = total - Max(booked, lastSecondBooked);
+    i16 budgetInt = static_cast<i16>(Max(budget, 0.0));
     if (budget < -0.1) {
         isStarvedPresent = true;
     }
+    for (size_t poolIdx = 0; poolIdx < Pools.size(); ++poolIdx) {
+        TPoolInfo& pool = Pools[poolIdx];
+        AtomicSet(pool.PotentialMaxThreadCount, Min(pool.MaxThreadCount, budgetInt));
+    }
     double overbooked = consumed - booked;
     if (isStarvedPresent) {
-      // last_starved_at_consumed_value = сумма по всем пулам consumed;
-      // TODO(cthulhu): использовать как лимит планвно устремлять этот лимит к total,
-      // использовать вместо total
-      if (beingStopped && beingStopped >= overbooked) {
-          // do nothing
-      } else {
-          TStackVec<size_t> reorder;
-          for (size_t i = 0; i < Pools.size(); ++i) {
-              reorder.push_back(i);
-          }
-          for (ui16 poolIdx : PriorityOrder) {
-              TPoolInfo &pool = Pools[poolIdx];
-              i64 threadCount = pool.GetThreadCount();
-              if (threadCount > pool.DefaultThreadCount) {
-                  pool.SetThreadCount(threadCount - 1);
-                  AtomicIncrement(pool.DecreasingThreadsByStarvedState);
-                  overbooked--;
-                  LWPROBE(HarmonizeOperation, poolIdx, pool.Pool->GetName(), "decrease", threadCount - 1, pool.DefaultThreadCount, pool.MaxThreadCount);
-                  if (overbooked < 1) {
-                      break;
-                  }
-              }
-          }
-      }
+        // last_starved_at_consumed_value = сумма по всем пулам consumed;
+        // TODO(cthulhu): использовать как лимит планвно устремлять этот лимит к total,
+        // использовать вместо total
+        if (beingStopped && beingStopped >= overbooked) {
+            // do nothing
+        } else {
+            TStackVec<size_t> reorder;
+            for (size_t i = 0; i < Pools.size(); ++i) {
+                reorder.push_back(i);
+            }
+            for (ui16 poolIdx : PriorityOrder) {
+                TPoolInfo &pool = Pools[poolIdx];
+                i64 threadCount = pool.GetThreadCount();
+                while (threadCount > pool.DefaultThreadCount) {
+                    pool.SetThreadCount(threadCount - 1);
+                    AtomicIncrement(pool.DecreasingThreadsByStarvedState);
+                    overbooked--;
+                    LWPROBE(HarmonizeOperation, poolIdx, pool.Pool->GetName(), "decrease", threadCount - 1, pool.DefaultThreadCount, pool.MaxThreadCount);
+                    if (overbooked < 1) {
+                        break;
+                    }
+                }
+                if (overbooked < 1) {
+                    break;
+                }
+            }
+        }
     } else {
         for (size_t needyPoolIdx : needyPools) {
             TPoolInfo &pool = Pools[needyPoolIdx];
@@ -422,6 +432,7 @@ TPoolHarmonizedStats THarmonizer::GetPoolStats(i16 poolId) const {
         .IncreasingThreadsByNeedyState = static_cast<ui64>(RelaxedLoad(&pool.IncreasingThreadsByNeedyState)),
         .DecreasingThreadsByStarvedState = static_cast<ui64>(RelaxedLoad(&pool.DecreasingThreadsByStarvedState)),
         .DecreasingThreadsByHoggishState = static_cast<ui64>(RelaxedLoad(&pool.DecreasingThreadsByHoggishState)),
+        .PotentialMaxThreadCount = static_cast<i16>(RelaxedLoad(&pool.PotentialMaxThreadCount)),
         .IsNeedy = static_cast<bool>(flags & 1),
         .IsStarved = static_cast<bool>(flags & 2),
         .IsHoggish = static_cast<bool>(flags & 4),