Refactor harmonizer in actorsystem (#12084)

Author: kruall

ydb/library/actors/core/cpu_manager.cpp

@@ -37,8 +37,8 @@ namespace NActors {
  poolsWithSharedThreads.push_back(cfg.PoolId);
  }
  }
- Shared.reset(new TSharedExecutorPool(Config.Shared, ExecutorPoolCount, poolsWithSharedThreads));
- auto sharedPool = static_cast<TSharedExecutorPool*>(Shared.get());
+ Shared.reset(CreateSharedExecutorPool(Config.Shared, ExecutorPoolCount, poolsWithSharedThreads));
+ auto sharedPool = static_cast<ISharedExecutorPool*>(Shared.get());
 
  ui64 ts = GetCycleCountFast();
  Harmonizer.reset(MakeHarmonizer(ts));
@@ -135,11 +135,9 @@ namespace NActors {
  for (TBasicExecutorPoolConfig& cfg : Config.Basic) {
  if (cfg.PoolId == poolId) {
  if (cfg.HasSharedThread) {
- auto *sharedPool = static_cast<TSharedExecutorPool*>(Shared.get());
+ auto *sharedPool = Shared.get();
  auto *pool = new TBasicExecutorPool(cfg, Harmonizer.get(), Jail.get());
- if (pool) {
- pool->AddSharedThread(sharedPool->GetSharedThread(poolId));
- }
+ pool->AddSharedThread(sharedPool->GetSharedThread(poolId));
  return pool;
  } else {
  return new TBasicExecutorPool(cfg, Harmonizer.get(), Jail.get());

ydb/library/actors/core/cpu_manager.h

@@ -2,10 +2,10 @@
 
 #include "config.h"
 #include "executor_pool_jail.h"
-#include "harmonizer.h"
 #include "executor_pool.h"
 #include "executor_pool_shared.h"
 #include "mon_stats.h"
+#include <ydb/library/actors/core/harmonizer/harmonizer.h>
 #include <memory>
 
 namespace NActors {
@@ -16,7 +16,7 @@ namespace NActors {
  const ui32 ExecutorPoolCount;
  TArrayHolder<TAutoPtr<IExecutorPool>> Executors;
  std::unique_ptr<IHarmonizer> Harmonizer;
- std::unique_ptr<TSharedExecutorPool> Shared;
+ std::unique_ptr<ISharedExecutorPool> Shared;
  std::unique_ptr<TExecutorPoolJail> Jail;
  TCpuManagerConfig Config;
 

ydb/library/actors/core/executor_pool.h

@@ -14,15 +14,16 @@ namespace NActors {
  struct TExecutorThreadStats;
  class TExecutorPoolJail;
  class ISchedulerCookie;
+ struct TSharedExecutorThreadCtx;
 
  struct TCpuConsumption {
- double ConsumedUs = 0;
- double BookedUs = 0;
+ double CpuUs = 0;
+ double ElapsedUs = 0;
  ui64 NotEnoughCpuExecutions = 0;
 
  void Add(const TCpuConsumption& other) {
- ConsumedUs += other.ConsumedUs;
- BookedUs += other.BookedUs;
+ CpuUs += other.CpuUs;
+ ElapsedUs += other.ElapsedUs;
  NotEnoughCpuExecutions += other.NotEnoughCpuExecutions;
  }
  };
@@ -176,6 +177,16 @@ namespace NActors {
  return 1;
  }
 
+ virtual TSharedExecutorThreadCtx* ReleaseSharedThread() {
+ return nullptr;
+ }
+ virtual void AddSharedThread(TSharedExecutorThreadCtx*) {
+ }
+
+ virtual bool IsSharedThreadEnabled() const {
+ return false;
+ }
+
  virtual TCpuConsumption GetThreadCpuConsumption(i16 threadIdx) {
  Y_UNUSED(threadIdx);
  return TCpuConsumption{0.0, 0.0};

ydb/library/actors/core/executor_pool_base.cpp

@@ -1,4 +1,5 @@
 #include "actorsystem.h"
+#include "activity_guard.h"
 #include "actor.h"
 #include "executor_pool_base.h"
 #include "executor_pool_basic_feature_flags.h"

ydb/library/actors/core/executor_pool_basic.cpp

@@ -407,10 +407,10 @@ namespace NActors {
  poolStats.DecreasingThreadsByHoggishState = stats.DecreasingThreadsByHoggishState;
  poolStats.DecreasingThreadsByExchange = stats.DecreasingThreadsByExchange;
  poolStats.PotentialMaxThreadCount = stats.PotentialMaxThreadCount;
- poolStats.MaxConsumedCpuUs = stats.MaxConsumedCpu;
- poolStats.MinConsumedCpuUs = stats.MinConsumedCpu;
- poolStats.MaxBookedCpuUs = stats.MaxBookedCpu;
- poolStats.MinBookedCpuUs = stats.MinBookedCpu;
+ poolStats.MaxCpuUs = stats.MaxCpuUs;
+ poolStats.MinCpuUs = stats.MinCpuUs;
+ poolStats.MaxElapsedUs = stats.MaxElapsedUs;
+ poolStats.MinElapsedUs = stats.MinElapsedUs;
  }
 
  statsCopy.resize(MaxFullThreadCount + 1);
@@ -429,7 +429,7 @@ namespace NActors {
  void TBasicExecutorPool::GetExecutorPoolState(TExecutorPoolState &poolState) const {
  if (Harmonizer) {
  TPoolHarmonizerStats stats = Harmonizer->GetPoolStats(PoolId);
- poolState.UsedCpu = stats.AvgConsumedCpu;
+ poolState.UsedCpu = stats.AvgElapsedUs;
  poolState.PossibleMaxLimit = stats.PotentialMaxThreadCount;
  } else {
  poolState.PossibleMaxLimit = poolState.MaxLimit;
@@ -625,7 +625,7 @@ namespace NActors {
  TExecutorThreadCtx& threadCtx = Threads[threadIdx];
  TExecutorThreadStats stats;
  threadCtx.Thread->GetCurrentStatsForHarmonizer(stats);
- return {Ts2Us(stats.SafeElapsedTicks), static_cast<double>(stats.CpuUs), stats.NotEnoughCpuExecutions};
+ return {static_cast<double>(stats.CpuUs), Ts2Us(stats.SafeElapsedTicks), stats.NotEnoughCpuExecutions};
  }
 
  i16 TBasicExecutorPool::GetBlockingThreadCount() const {

ydb/library/actors/core/executor_pool_basic.h

@@ -7,8 +7,8 @@
 #include "executor_pool_basic_feature_flags.h"
 #include "scheduler_queue.h"
 #include "executor_pool_base.h"
-#include "harmonizer.h"
 #include <memory>
+#include <ydb/library/actors/core/harmonizer/harmonizer.h>
 #include <ydb/library/actors/actor_type/indexes.h>
 #include <ydb/library/actors/util/unordered_cache.h>
 #include <ydb/library/actors/util/threadparkpad.h>
@@ -278,8 +278,11 @@ namespace NActors {
  void CalcSpinPerThread(ui64 wakingUpConsumption);
  void ClearWaitingStats() const;
 
- TSharedExecutorThreadCtx* ReleaseSharedThread();
- void AddSharedThread(TSharedExecutorThreadCtx* thread);
+ TSharedExecutorThreadCtx* ReleaseSharedThread() override;
+ void AddSharedThread(TSharedExecutorThreadCtx* thread) override;
+ bool IsSharedThreadEnabled() const override {
+ return true;
+ }
 
  private:
  void AskToGoToSleep(bool *needToWait, bool *needToBlock);

ydb/library/actors/core/executor_pool_io.cpp

@@ -156,7 +156,7 @@ namespace NActors {
  void TIOExecutorPool::GetExecutorPoolState(TExecutorPoolState &poolState) const {
  if (Harmonizer) {
  TPoolHarmonizerStats stats = Harmonizer->GetPoolStats(PoolId);
- poolState.UsedCpu = stats.AvgConsumedCpu;
+ poolState.UsedCpu = stats.AvgElapsedUs;
  }
  poolState.CurrentLimit = PoolThreads;
  poolState.MaxLimit = PoolThreads;

ydb/library/actors/core/executor_pool_io.h

@@ -3,9 +3,9 @@
 #include "actorsystem.h"
 #include "executor_thread.h"
 #include "executor_thread_ctx.h"
-#include "harmonizer.h"
 #include "scheduler_queue.h"
 #include "executor_pool_base.h"
+#include <ydb/library/actors/core/harmonizer/harmonizer.h>
 #include <ydb/library/actors/actor_type/indexes.h>
 #include <ydb/library/actors/util/ticket_lock.h>
 #include <ydb/library/actors/util/unordered_cache.h>

ydb/library/actors/core/executor_pool_shared.cpp

@@ -12,6 +12,50 @@
 
 namespace NActors {
 
+class TSharedExecutorPool: public ISharedExecutorPool {
+public:
+ TSharedExecutorPool(const TSharedExecutorPoolConfig &config, i16 poolCount, std::vector<i16> poolsWithThreads);
+
+ // IThreadPool
+ void Prepare(TActorSystem* actorSystem, NSchedulerQueue::TReader** scheduleReaders, ui32* scheduleSz) override;
+ void Start() override;
+ void PrepareStop() override;
+ void Shutdown() override;
+ bool Cleanup() override;
+
+ TSharedExecutorThreadCtx *GetSharedThread(i16 poolId) override;
+ void GetSharedStats(i16 pool, std::vector<TExecutorThreadStats>& statsCopy) override;
+ void GetSharedStatsForHarmonizer(i16 pool, std::vector<TExecutorThreadStats>& statsCopy) override;
+ TCpuConsumption GetThreadCpuConsumption(i16 poolId, i16 threadIdx) override;
+ std::vector<TCpuConsumption> GetThreadsCpuConsumption(i16 poolId) override;
+
+ i16 ReturnOwnHalfThread(i16 pool) override;
+ i16 ReturnBorrowedHalfThread(i16 pool) override;
+ void GiveHalfThread(i16 from, i16 to) override;
+
+ i16 GetSharedThreadCount() const override;
+
+ TSharedPoolState GetState() const override;
+
+ void Init(const std::vector<IExecutorPool*>& pools, bool withThreads) override;
+
+private:
+ TSharedPoolState State;
+
+ std::vector<IExecutorPool*> Pools;
+
+ i16 PoolCount;
+ i16 SharedThreadCount;
+ std::unique_ptr<TSharedExecutorThreadCtx[]> Threads;
+
+ std::unique_ptr<NSchedulerQueue::TReader[]> ScheduleReaders;
+ std::unique_ptr<NSchedulerQueue::TWriter[]> ScheduleWriters;
+
+ TDuration TimePerMailbox;
+ ui32 EventsPerMailbox;
+ ui64 SoftProcessingDurationTs;
+}; // class TSharedExecutorPool
+
 TSharedExecutorPool::TSharedExecutorPool(const TSharedExecutorPoolConfig &config, i16 poolCount, std::vector<i16> poolsWithThreads)
  : State(poolCount, poolsWithThreads.size())
  , Pools(poolCount)
@@ -29,40 +73,47 @@ TSharedExecutorPool::TSharedExecutorPool(const TSharedExecutorPoolConfig &config
  }
 }
 
-void TSharedExecutorPool::Prepare(TActorSystem* actorSystem, NSchedulerQueue::TReader** scheduleReaders, ui32* scheduleSz) {
- // ActorSystem = actorSystem;
-
- ScheduleReaders.reset(new NSchedulerQueue::TReader[SharedThreadCount]);
- ScheduleWriters.reset(new NSchedulerQueue::TWriter[SharedThreadCount]);
-
- std::vector<IExecutorPool*> poolsBasic = actorSystem->GetBasicExecutorPools();
- std::vector<IExecutorPool*> poolByThread(SharedThreadCount);
- for (IExecutorPool* pool : poolsBasic) {
- Pools[pool->PoolId] = dynamic_cast<TBasicExecutorPool*>(pool);
- i16 threadIdx = State.ThreadByPool[pool->PoolId];
- if (threadIdx >= 0) {
- poolByThread[threadIdx] = pool;
- }
+void TSharedExecutorPool::Init(const std::vector<IExecutorPool*>& pools, bool withThreads) {
+ std::vector<IExecutorPool*> poolByThread(SharedThreadCount);
+ for (IExecutorPool* pool : pools) {
+ Pools[pool->PoolId] = pool;
+ i16 threadIdx = State.ThreadByPool[pool->PoolId];
+ if (threadIdx >= 0) {
+ poolByThread[threadIdx] = pool;
  }
+ }
 
- for (i16 i = 0; i != SharedThreadCount; ++i) {
- // !TODO
- Threads[i].ExecutorPools[0].store(dynamic_cast<TBasicExecutorPool*>(poolByThread[i]), std::memory_order_release);
+ for (i16 i = 0; i != SharedThreadCount; ++i) {
+ // !TODO
+ Threads[i].ExecutorPools[0].store(dynamic_cast<TBasicExecutorPool*>(poolByThread[i]), std::memory_order_release);
+ if (withThreads) {
  Threads[i].Thread.reset(
  new TSharedExecutorThread(
  -1,
-  actorSystem,
-  &Threads[i],
-  PoolCount,
-  "SharedThread",
-  SoftProcessingDurationTs,
-  TimePerMailbox,
+ nullptr,
+ &Threads[i],
+ PoolCount,
+ "SharedThread",
+ SoftProcessingDurationTs,
+ TimePerMailbox,
  EventsPerMailbox));
- ScheduleWriters[i].Init(ScheduleReaders[i]);
  }
+ }
+}
+
+void TSharedExecutorPool::Prepare(TActorSystem* actorSystem, NSchedulerQueue::TReader** scheduleReaders, ui32* scheduleSz) {
+ ScheduleReaders.reset(new NSchedulerQueue::TReader[SharedThreadCount]);
+ ScheduleWriters.reset(new NSchedulerQueue::TWriter[SharedThreadCount]);
+
+ std::vector<IExecutorPool*> poolsBasic = actorSystem->GetBasicExecutorPools();
+ Init(poolsBasic, true);
 
- *scheduleReaders = ScheduleReaders.get();
- *scheduleSz = SharedThreadCount;
+ for (i16 i = 0; i != SharedThreadCount; ++i) {
+ ScheduleWriters[i].Init(ScheduleReaders[i]);
+ }
+
+ *scheduleReaders = ScheduleReaders.get();
+ *scheduleSz = SharedThreadCount;
 }
 
 void TSharedExecutorPool::Start() {
@@ -99,24 +150,27 @@ TSharedExecutorThreadCtx* TSharedExecutorPool::GetSharedThread(i16 pool) {
  return &Threads[threadIdx];
 }
 
-void TSharedExecutorPool::ReturnOwnHalfThread(i16 pool) {
+i16 TSharedExecutorPool::ReturnOwnHalfThread(i16 pool) {
  i16 threadIdx = State.ThreadByPool[pool];
- TBasicExecutorPool* borrowingPool = Threads[threadIdx].ExecutorPools[1].exchange(nullptr, std::memory_order_acq_rel);
+ IExecutorPool* borrowingPool = Threads[threadIdx].ExecutorPools[1].exchange(nullptr, std::memory_order_acq_rel);
  Y_ABORT_UNLESS(borrowingPool);
- State.BorrowedThreadByPool[State.PoolByBorrowedThread[threadIdx]] = -1;
+ i16 borrowedPool = State.PoolByBorrowedThread[threadIdx];
+ State.BorrowedThreadByPool[borrowedPool] = -1;
  State.PoolByBorrowedThread[threadIdx] = -1;
  // TODO(kruall): Check on race
  borrowingPool->ReleaseSharedThread();
+ return borrowedPool;
 }
 
-void TSharedExecutorPool::ReturnBorrowedHalfThread(i16 pool) {
+i16 TSharedExecutorPool::ReturnBorrowedHalfThread(i16 pool) {
  i16 threadIdx = State.BorrowedThreadByPool[pool];
- TBasicExecutorPool* borrowingPool = Threads[threadIdx].ExecutorPools[1].exchange(nullptr, std::memory_order_acq_rel);
+ IExecutorPool* borrowingPool = Threads[threadIdx].ExecutorPools[1].exchange(nullptr, std::memory_order_acq_rel);
  Y_ABORT_UNLESS(borrowingPool);
  State.BorrowedThreadByPool[State.PoolByBorrowedThread[threadIdx]] = -1;
  State.PoolByBorrowedThread[threadIdx] = -1;
  // TODO(kruall): Check on race
  borrowingPool->ReleaseSharedThread();
+ return State.PoolByThread[threadIdx];
 }
 
 void TSharedExecutorPool::GiveHalfThread(i16 from, i16 to) {
@@ -127,14 +181,14 @@ void TSharedExecutorPool::GiveHalfThread(i16 from, i16 to) {
  if (borrowedThreadIdx != -1) {
  i16 originalPool = State.PoolByThread[borrowedThreadIdx];
  if (originalPool == to) {
- return ReturnOwnHalfThread(to);
+ ReturnOwnHalfThread(to);
  } else {
  ReturnOwnHalfThread(originalPool);
  }
  from = originalPool;
  }
  i16 threadIdx = State.ThreadByPool[from];
- TBasicExecutorPool* borrowingPool = Pools[to];
+ IExecutorPool* borrowingPool = Pools[to];
  Threads[threadIdx].ExecutorPools[1].store(borrowingPool, std::memory_order_release);
  State.BorrowedThreadByPool[to] = threadIdx;
  State.PoolByBorrowedThread[threadIdx] = to;
@@ -143,16 +197,16 @@ void TSharedExecutorPool::GiveHalfThread(i16 from, i16 to) {
 }
 
 void TSharedExecutorPool::GetSharedStats(i16 poolId, std::vector<TExecutorThreadStats>& statsCopy) {
- statsCopy.resize(SharedThreadCount + 1);
+ statsCopy.resize(SharedThreadCount);
  for (i16 i = 0; i < SharedThreadCount; ++i) {
- Threads[i].Thread->GetSharedStats(poolId, statsCopy[i + 1]);
+ Threads[i].Thread->GetSharedStats(poolId, statsCopy[i]);
  }
 }
 
 void TSharedExecutorPool::GetSharedStatsForHarmonizer(i16 poolId, std::vector<TExecutorThreadStats>& statsCopy) {
- statsCopy.resize(SharedThreadCount + 1);
+ statsCopy.resize(SharedThreadCount);
  for (i16 i = 0; i < SharedThreadCount; ++i) {
- Threads[i].Thread->GetSharedStatsForHarmonizer(poolId, statsCopy[i + 1]);
+ Threads[i].Thread->GetSharedStatsForHarmonizer(poolId, statsCopy[i]);
  }
 }
 
@@ -181,4 +235,34 @@ TSharedPoolState TSharedExecutorPool::GetState() const {
  return State;
 }
 
+ISharedExecutorPool *CreateSharedExecutorPool(const TSharedExecutorPoolConfig &config, i16 poolCount, std::vector<i16> poolsWithThreads) {
+ return new TSharedExecutorPool(config, poolCount, poolsWithThreads);
+}
+
+TString TSharedPoolState::ToString() const {
+ TStringBuilder builder;
+ builder << '{';
+ builder << "ThreadByPool: [";
+ for (ui32 i = 0; i < ThreadByPool.size(); ++i) {
+ builder << ThreadByPool[i] << (i == ThreadByPool.size() - 1 ? "" : ", ");
+ }
+ builder << "], ";
+ builder << "PoolByThread: [";
+ for (ui32 i = 0; i < PoolByThread.size(); ++i) {
+ builder << PoolByThread[i] << (i == PoolByThread.size() - 1 ? "" : ", ");
+ }
+ builder << "], ";
+ builder << "BorrowedThreadByPool: [";
+ for (ui32 i = 0; i < BorrowedThreadByPool.size(); ++i) {
+ builder << BorrowedThreadByPool[i] << (i == BorrowedThreadByPool.size() - 1 ? "" : ", ");
+ }
+ builder << "], ";
+ builder << "PoolByBorrowedThread: [";
+ for (ui32 i = 0; i < PoolByBorrowedThread.size(); ++i) {
+ builder << PoolByBorrowedThread[i] << (i == PoolByBorrowedThread.size() - 1 ? "" : ", ");
+ }
+ builder << ']';
+ return builder << '}';
+}
+
 }