Fix false starving state,

9 files changed, 24 insertions, 31 deletions

diff --git a/library/cpp/actors/core/executor_pool.h b/library/cpp/actors/core/executor_pool.h
index f39415c7e2e..c7c85e61fd2 100644
--- a/library/cpp/actors/core/executor_pool.h
+++ b/library/cpp/actors/core/executor_pool.h
@@ -10,6 +10,11 @@ namespace NActors {
     struct TWorkerContext;
     class ISchedulerCookie;
 
+    struct TCpuConsumption {
+        double ConsumedUs = 0;
+        double BookedUs = 0;
+    };
+
     class IExecutorPool : TNonCopyable {
     public:
         const ui32 PoolId;
@@ -131,14 +136,9 @@ namespace NActors {
             return false;
         }
 
-        virtual double GetThreadConsumedUs(i16 threadIdx) {
-            Y_UNUSED(threadIdx);
-            return 0.0;
-        }
-
-        virtual double GetThreadBookedUs(i16 threadIdx) {
+        virtual TCpuConsumption GetThreadCpuConsumption(i16 threadIdx) {
             Y_UNUSED(threadIdx);
-            return 0.0;
+            return TCpuConsumption{0.0, 0.0};
         }
     };
 
diff --git a/library/cpp/actors/core/executor_pool_basic.cpp b/library/cpp/actors/core/executor_pool_basic.cpp
index a0f5d4423c3..de04105991d 100644
--- a/library/cpp/actors/core/executor_pool_basic.cpp
+++ b/library/cpp/actors/core/executor_pool_basic.cpp
@@ -493,24 +493,14 @@ namespace NActors {
         return false;
     }
 
-    double TBasicExecutorPool::GetThreadConsumedUs(i16 threadIdx) {
+    TCpuConsumption TBasicExecutorPool::GetThreadCpuConsumption(i16 threadIdx) {
         if ((ui32)threadIdx >= PoolThreads) {
-            return 0;
+            return {0.0, 0.0};
         }
         TThreadCtx& threadCtx = Threads[threadIdx];
         TExecutorThreadStats stats;
         threadCtx.Thread->GetCurrentStats(stats);
-        return Ts2Us(stats.ElapsedTicks);
-    }
-
-    double TBasicExecutorPool::GetThreadBookedUs(i16 threadIdx) {
-        if ((ui32)threadIdx >= PoolThreads) {
-            return 0;
-        }
-        TThreadCtx& threadCtx = Threads[threadIdx];
-        TExecutorThreadStats stats;
-        threadCtx.Thread->GetCurrentStats(stats);
-        return stats.CpuNs / 1000.0;
+        return {Ts2Us(stats.SafeElapsedTicks), static_cast<double>(stats.CpuUs)};
     }
 
     i16 TBasicExecutorPool::GetBlockingThreadCount() const {
diff --git a/library/cpp/actors/core/executor_pool_basic.h b/library/cpp/actors/core/executor_pool_basic.h
index cd94a998f1a..813f91dc9a9 100644
--- a/library/cpp/actors/core/executor_pool_basic.h
+++ b/library/cpp/actors/core/executor_pool_basic.h
@@ -153,8 +153,7 @@ namespace NActors {
         i16 GetMinThreadCount() const override;
         i16 GetMaxThreadCount() const override;
         bool IsThreadBeingStopped(i16 threadIdx) const override;
-        double GetThreadConsumedUs(i16 threadIdx) override;
-        double GetThreadBookedUs(i16 threadIdx) override;
+        TCpuConsumption GetThreadCpuConsumption(i16 threadIdx) override;
         i16 GetBlockingThreadCount() const override;
         i16 GetPriority() const override;
 
diff --git a/library/cpp/actors/core/executor_pool_basic_ut.cpp b/library/cpp/actors/core/executor_pool_basic_ut.cpp
index 6361bc66626..f96f65931a0 100644
--- a/library/cpp/actors/core/executor_pool_basic_ut.cpp
+++ b/library/cpp/actors/core/executor_pool_basic_ut.cpp
@@ -339,7 +339,7 @@ Y_UNIT_TEST_SUITE(BasicExecutorPool) {
         UNIT_ASSERT_VALUES_EQUAL(stats[0].PreemptedEvents, 0);
         UNIT_ASSERT_VALUES_EQUAL(stats[0].NonDeliveredEvents, 0);
         UNIT_ASSERT_VALUES_EQUAL(stats[0].EmptyMailboxActivation, 0);
-        //UNIT_ASSERT_VALUES_EQUAL(stats[0].CpuNs, 0); // depends on total duration of test, so undefined
+        //UNIT_ASSERT_VALUES_EQUAL(stats[0].CpuUs, 0); // depends on total duration of test, so undefined
         UNIT_ASSERT(stats[0].ElapsedTicks > 0);
         UNIT_ASSERT(stats[0].ParkedTicks > 0);
         UNIT_ASSERT_VALUES_EQUAL(stats[0].BlockedTicks, 0);
diff --git a/library/cpp/actors/core/executor_pool_united_ut.cpp b/library/cpp/actors/core/executor_pool_united_ut.cpp
index 133e9c5f2aa..a7c7399d73e 100644
--- a/library/cpp/actors/core/executor_pool_united_ut.cpp
+++ b/library/cpp/actors/core/executor_pool_united_ut.cpp
@@ -171,7 +171,7 @@ Y_UNIT_TEST_SUITE(UnitedExecutorPool) {
         //UNIT_ASSERT_VALUES_EQUAL(stats[0].PreemptedEvents, 0); // depends on execution time and system load, so may be non-zero
         UNIT_ASSERT_VALUES_EQUAL(stats[0].NonDeliveredEvents, 0);
         UNIT_ASSERT_VALUES_EQUAL(stats[0].EmptyMailboxActivation, 0);
-        //UNIT_ASSERT_VALUES_EQUAL(stats[0].CpuNs, 0); // depends on total duration of test, so undefined
+        //UNIT_ASSERT_VALUES_EQUAL(stats[0].CpuUs, 0); // depends on total duration of test, so undefined
         UNIT_ASSERT(stats[0].ElapsedTicks > 0);
         //UNIT_ASSERT(stats[0].ParkedTicks == 0); // per-pool parked time does not make sense for united pools
         UNIT_ASSERT_VALUES_EQUAL(stats[0].BlockedTicks, 0);
diff --git a/library/cpp/actors/core/harmonizer.cpp b/library/cpp/actors/core/harmonizer.cpp
index 17f00f09c6e..1b547a9ad6e 100644
--- a/library/cpp/actors/core/harmonizer.cpp
+++ b/library/cpp/actors/core/harmonizer.cpp
@@ -170,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));
     }
 }
@@ -237,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) {
diff --git a/library/cpp/actors/core/mon_stats.h b/library/cpp/actors/core/mon_stats.h
index 0fcbf6fcf4c..4c664a964a4 100644
--- a/library/cpp/actors/core/mon_stats.h
+++ b/library/cpp/actors/core/mon_stats.h
@@ -79,7 +79,8 @@ namespace NActors {
         ui64 PreemptedEvents = 0; // Number of events experienced hard preemption
         ui64 NonDeliveredEvents = 0;
         ui64 EmptyMailboxActivation = 0;
-        ui64 CpuNs = 0; // nanoseconds thread was executing on CPU (accounts for preemtion)
+        ui64 CpuUs = 0; // microseconds thread was executing on CPU (accounts for preemtion)
+        ui64 SafeElapsedTicks = 0;
         ui64 WorstActivationTimeUs = 0;
         NHPTimer::STime ElapsedTicks = 0;
         NHPTimer::STime ParkedTicks = 0;
@@ -123,7 +124,8 @@ namespace NActors {
             PreemptedEvents += RelaxedLoad(&other.PreemptedEvents);
             NonDeliveredEvents += RelaxedLoad(&other.NonDeliveredEvents);
             EmptyMailboxActivation += RelaxedLoad(&other.EmptyMailboxActivation);
-            CpuNs += RelaxedLoad(&other.CpuNs);
+            CpuUs += RelaxedLoad(&other.CpuUs);
+            SafeElapsedTicks += RelaxedLoad(&other.SafeElapsedTicks);
             RelaxedStore(
                 &WorstActivationTimeUs,
                 std::max(RelaxedLoad(&WorstActivationTimeUs), RelaxedLoad(&other.WorstActivationTimeUs)));
diff --git a/library/cpp/actors/core/worker_context.h b/library/cpp/actors/core/worker_context.h
index 2179771fb69..c3a2947df19 100644
--- a/library/cpp/actors/core/worker_context.h
+++ b/library/cpp/actors/core/worker_context.h
@@ -137,7 +137,8 @@ namespace NActors {
         }
 
         void UpdateThreadTime() {
-            RelaxedStore(&WorkerStats.CpuNs, ThreadCPUTime() * 1000);
+            RelaxedStore(&WorkerStats.SafeElapsedTicks, (ui64)RelaxedLoad(&WorkerStats.ElapsedTicks));
+            RelaxedStore(&WorkerStats.CpuUs, ThreadCPUTime());
         }
 #else
         void GetCurrentStats(TExecutorThreadStats&) const {}
diff --git a/library/cpp/actors/helpers/pool_stats_collector.h b/library/cpp/actors/helpers/pool_stats_collector.h
index a56b4d9799f..d80951827d0 100644
--- a/library/cpp/actors/helpers/pool_stats_collector.h
+++ b/library/cpp/actors/helpers/pool_stats_collector.h
@@ -217,7 +217,7 @@ private:
             *NonDeliveredEvents = stats.NonDeliveredEvents;
             *DestroyedActors    = stats.PoolDestroyedActors;
             *EmptyMailboxActivation = stats.EmptyMailboxActivation;
-            *CpuMicrosec        = stats.CpuNs / 1000;
+            *CpuMicrosec        = stats.CpuUs;
             *ElapsedMicrosec    = ::NHPTimer::GetSeconds(stats.ElapsedTicks)*1000000;
             *ParkedMicrosec     = ::NHPTimer::GetSeconds(stats.ParkedTicks)*1000000;
             *ActorRegistrations = stats.PoolActorRegistrations;