Improve harmonizer on pulling stats (#5170)

Author: kruall

ydb/library/actors/core/executor_pool_basic.cpp

@@ -610,7 +610,7 @@ namespace NActors {
         }
         TExecutorThreadCtx& threadCtx = Threads[threadIdx];
         TExecutorThreadStats stats;
-        threadCtx.Thread->GetCurrentStats(stats);
+        threadCtx.Thread->GetCurrentStatsForHarmonizer(stats);
         return {Ts2Us(stats.SafeElapsedTicks), static_cast<double>(stats.CpuUs), stats.NotEnoughCpuExecutions};
     }
 

ydb/library/actors/core/executor_pool_shared.cpp

@@ -149,12 +149,19 @@ void TSharedExecutorPool::GetSharedStats(i16 poolId, std::vector<TExecutorThread
     }
 }
 
+void TSharedExecutorPool::GetSharedStatsForHarmonizer(i16 poolId, std::vector<TExecutorThreadStats>& statsCopy) {
+    statsCopy.resize(SharedThreadCount + 1);
+    for (i16 i = 0; i < SharedThreadCount; ++i) {
+        Threads[i].Thread->GetSharedStatsForHarmonizer(poolId, statsCopy[i + 1]);
+    }
+}
+
 TCpuConsumption TSharedExecutorPool::GetThreadCpuConsumption(i16 poolId, i16 threadIdx) {
     if (threadIdx >= SharedThreadCount) {
         return {0.0, 0.0};
     }
     TExecutorThreadStats stats;
-    Threads[threadIdx].Thread->GetSharedStats(poolId, stats);
+    Threads[threadIdx].Thread->GetSharedStatsForHarmonizer(poolId, stats);
     return {Ts2Us(stats.SafeElapsedTicks), static_cast<double>(stats.CpuUs), stats.NotEnoughCpuExecutions};
 }
 

ydb/library/actors/core/executor_pool_shared.h

@@ -37,6 +37,7 @@ namespace NActors {
 
         TSharedExecutorThreadCtx *GetSharedThread(i16 poolId);
         void GetSharedStats(i16 pool, std::vector<TExecutorThreadStats>& statsCopy);
+        void GetSharedStatsForHarmonizer(i16 pool, std::vector<TExecutorThreadStats>& statsCopy);
         TCpuConsumption GetThreadCpuConsumption(i16 poolId, i16 threadIdx);
         std::vector<TCpuConsumption> GetThreadsCpuConsumption(i16 poolId);
 

ydb/library/actors/core/executor_thread.cpp

@@ -811,6 +811,19 @@ namespace NActors {
         statsCopy.Aggregate(SharedStats[poolId]);
     }
 
+    void TGenericExecutorThread::GetCurrentStatsForHarmonizer(TExecutorThreadStats& statsCopy) {
+        statsCopy.SafeElapsedTicks = RelaxedLoad(&Ctx.Stats->SafeElapsedTicks);
+        statsCopy.CpuUs = RelaxedLoad(&Ctx.Stats->CpuUs);
+        statsCopy.NotEnoughCpuExecutions = RelaxedLoad(&Ctx.Stats->NotEnoughCpuExecutions);
+    }
+
+    void TGenericExecutorThread::GetSharedStatsForHarmonizer(i16 poolId, TExecutorThreadStats &stats) {
+        stats.SafeElapsedTicks = RelaxedLoad(&SharedStats[poolId].SafeElapsedTicks);
+        stats.CpuUs = RelaxedLoad(&SharedStats[poolId].CpuUs);
+        stats.NotEnoughCpuExecutions = RelaxedLoad(&SharedStats[poolId].NotEnoughCpuExecutions);
+    }
+
+
     TGenericExecutorThreadCtx::~TGenericExecutorThreadCtx()
     {}
 }

ydb/library/actors/core/executor_thread.h

@@ -71,6 +71,9 @@ namespace NActors {
         void GetCurrentStats(TExecutorThreadStats& statsCopy);
         void GetSharedStats(i16 poolId, TExecutorThreadStats &stats);
 
+        void GetCurrentStatsForHarmonizer(TExecutorThreadStats& statsCopy);
+        void GetSharedStatsForHarmonizer(i16 poolId, TExecutorThreadStats &stats);
+
         TThreadId GetThreadId() const; // blocks, must be called after Start()
         TWorkerId GetWorkerId() const;
 

ydb/library/actors/core/harmonizer.cpp

@@ -295,7 +295,7 @@ TCpuConsumption TPoolInfo::PullStats(ui64 ts) {
     }
     TVector<TExecutorThreadStats> sharedStats;
     if (Shared) {
-        Shared->GetSharedStats(Pool->PoolId, sharedStats);
+        Shared->GetSharedStatsForHarmonizer(Pool->PoolId, sharedStats);
     }
 
     for (ui32 sharedIdx = 0; sharedIdx < SharedInfo.size(); ++sharedIdx) {