Add local queue configs (#14802)

kruall · web-flow · commit 2992756a5c08 · 2025-02-20T14:32:08.000+03:00
diff --git a/ydb/core/driver_lib/run/auto_config_initializer.cpp b/ydb/core/driver_lib/run/auto_config_initializer.cpp
@@ -328,6 +328,13 @@ namespace NKikimr::NAutoConfigInitializer {
             } else {
                 executor->SetSpinThreshold(1);
             }
+
+            if (config->HasMinLocalQueueSize()) {
+                executor->SetMinLocalQueueSize(config->GetMinLocalQueueSize());
+            }
+            if (config->HasMaxLocalQueueSize()) {
+                executor->SetMaxLocalQueueSize(config->GetMaxLocalQueueSize());
+            }
         }
     }
 
diff --git a/ydb/core/driver_lib/run/config_helpers.cpp b/ydb/core/driver_lib/run/config_helpers.cpp
@@ -86,7 +86,14 @@ void AddExecutorPool(NActors::TCpuManagerConfig& cpuManager, const NKikimrConfig
             basic.MaxThreadCount = poolConfig.GetMaxThreads();
             basic.DefaultThreadCount = poolConfig.GetThreads();
             basic.Priority = poolConfig.GetPriority();
+            if (poolConfig.HasMinLocalQueueSize()) {
+                basic.MinLocalQueueSize = poolConfig.GetMinLocalQueueSize();
+            }
+            if (poolConfig.HasMaxLocalQueueSize()) {
+                basic.MaxLocalQueueSize = poolConfig.GetMaxLocalQueueSize();
+            }
             cpuManager.Basic.emplace_back(std::move(basic));
+            
             break;
         }
 
diff --git a/ydb/core/protos/config.proto b/ydb/core/protos/config.proto
@@ -86,6 +86,8 @@ message TActorSystemConfig {
         optional int32 MaxAvgPingDeviation = 17;
 
         optional bool HasSharedThread = 18;
+        optional uint32 MaxLocalQueueSize = 20;
+        optional uint32 MinLocalQueueSize = 21;
     }
 
     message TScheduler {
@@ -125,6 +127,9 @@ message TActorSystemConfig {
 
     optional bool MonitorStuckActors = 15;
     optional EActorSystemProfile ActorSystemProfile = 16;
+
+    optional uint32 MinLocalQueueSize = 20;
+    optional uint32 MaxLocalQueueSize = 21;
 }
 
 message TStaticNameserviceConfig {
diff --git a/ydb/library/actors/core/config.h b/ydb/library/actors/core/config.h
@@ -32,6 +32,8 @@ namespace NActors {
         EASProfile ActorSystemProfile = EASProfile::Default;
         bool HasSharedThread = false;
         bool UseRingQueue = false;
+        ui16 MinLocalQueueSize = 0;
+        ui16 MaxLocalQueueSize = 0;
     };
 
     struct TSharedExecutorPoolConfig {
diff --git a/ydb/library/actors/core/executor_pool_basic.cpp b/ydb/library/actors/core/executor_pool_basic.cpp
@@ -113,6 +113,8 @@ namespace NActors {
         , Harmonizer(harmonizer)
         , SoftProcessingDurationTs(cfg.SoftProcessingDurationTs)
         , HasOwnSharedThread(cfg.HasSharedThread)
+        , MaxLocalQueueSize(cfg.MaxLocalQueueSize)
+        , MinLocalQueueSize(cfg.MinLocalQueueSize)
         , Priority(cfg.Priority)
         , Jail(jail)
         , ActorSystemProfile(cfg.ActorSystemProfile)
@@ -124,13 +126,9 @@ namespace NActors {
             threads = threads - 1;
         }
 
-        if constexpr (NFeatures::IsLocalQueues()) {
+        if (MaxLocalQueueSize) {
             LocalQueues.Reset(new NThreading::TPadded<std::queue<ui32>>[threads]);
-            if constexpr (NFeatures::TLocalQueuesFeatureFlags::FIXED_LOCAL_QUEUE_SIZE) {
-                LocalQueueSize = *NFeatures::TLocalQueuesFeatureFlags::FIXED_LOCAL_QUEUE_SIZE;
-            } else {
-                LocalQueueSize = NFeatures::TLocalQueuesFeatureFlags::MIN_LOCAL_QUEUE_SIZE;
-            }
+            LocalQueueSize = MinLocalQueueSize;
         }
         if constexpr (NFeatures::TSpinFeatureFlags::CalcPerThread) {
             for (ui32 idx = 0; idx < threads; ++idx) {
@@ -284,7 +282,7 @@ namespace NActors {
     }
 
     TMailbox* TBasicExecutorPool::GetReadyActivation(ui64 revolvingCounter) {
-        if constexpr (NFeatures::IsLocalQueues()) {
+        if (MaxLocalQueueSize) {
             EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "local queue");
             return GetReadyActivationLocalQueue(revolvingCounter);
         } else {
@@ -389,7 +387,7 @@ namespace NActors {
     }
 
     void TBasicExecutorPool::ScheduleActivationEx(TMailbox* mailbox, ui64 revolvingCounter) {
-        if constexpr (NFeatures::IsLocalQueues()) {
+        if (MaxLocalQueueSize) {
             ScheduleActivationExLocalQueue(mailbox, revolvingCounter);
         } else {
             ScheduleActivationExCommon(mailbox, revolvingCounter, AtomicGet(Semaphore));
@@ -629,9 +627,8 @@ namespace NActors {
     }
 
     void TBasicExecutorPool::SetLocalQueueSize(ui16 size) {
-        if constexpr (!NFeatures::TLocalQueuesFeatureFlags::FIXED_LOCAL_QUEUE_SIZE) {
-            LocalQueueSize.store(std::max(size, NFeatures::TLocalQueuesFeatureFlags::MAX_LOCAL_QUEUE_SIZE), std::memory_order_relaxed);
-        }
+        size = std::min(size, MaxLocalQueueSize);
+        LocalQueueSize.store(size, std::memory_order_relaxed);
     }
 
     void TBasicExecutorPool::Initialize() {
@@ -753,4 +750,16 @@ namespace NActors {
         return EventsPerMailboxValue;
     }
 
+    ui16 TBasicExecutorPool::GetLocalQueueSize() const {
+        return LocalQueueSize.load(std::memory_order_relaxed);
+    }
+
+    ui16 TBasicExecutorPool::GetMaxLocalQueueSize() const {
+        return MaxLocalQueueSize;
+    }
+
+    ui16 TBasicExecutorPool::GetMinLocalQueueSize() const {
+        return MinLocalQueueSize;
+    }
+
 }
diff --git a/ydb/library/actors/core/executor_pool_basic.h b/ydb/library/actors/core/executor_pool_basic.h
@@ -174,6 +174,8 @@ namespace NActors {
         IHarmonizer *Harmonizer;
         ui64 SoftProcessingDurationTs = 0;
         bool HasOwnSharedThread = false;
+        ui16 MaxLocalQueueSize = 0;
+        ui16 MinLocalQueueSize = 0;
 
         const i16 Priority = 0;
         const ui32 ActorSystemIndex = NActors::TActorTypeOperator::GetActorSystemIndex();
@@ -243,7 +245,9 @@ namespace NActors {
         void ScheduleActivationExLocalQueue(TMailbox* mailbox, ui64 revolvingWriteCounter);
 
         void SetLocalQueueSize(ui16 size);
-
+        ui16 GetLocalQueueSize() const;
+        ui16 GetMaxLocalQueueSize() const;
+        ui16 GetMinLocalQueueSize() const;
         void Prepare(TActorSystem* actorSystem, NSchedulerQueue::TReader** scheduleReaders, ui32* scheduleSz) override;
         void Start() override;
         void PrepareStop() override;
diff --git a/ydb/library/actors/core/executor_pool_basic_feature_flags.h b/ydb/library/actors/core/executor_pool_basic_feature_flags.h
@@ -21,10 +21,6 @@ namespace NActors::NFeatures {
     struct TLocalQueuesFeatureFlags {
         static constexpr EActorSystemOptimizationType OptimizationType = EActorSystemOptimizationType::LocalQueues;
 
-        static constexpr ui16 MIN_LOCAL_QUEUE_SIZE = 0;
-        static constexpr ui16 MAX_LOCAL_QUEUE_SIZE = 16;
-        static constexpr std::optional<ui16> FIXED_LOCAL_QUEUE_SIZE = std::nullopt;
-
         static constexpr bool UseIfAllOtherThreadsAreSleeping = false;
         static constexpr bool UseOnMicroburst = false;
     };
@@ -43,8 +39,4 @@ namespace NActors::NFeatures {
         return TFeatureFlags::OptimizationType == EActorSystemOptimizationType::Common;
     }
 
-    consteval bool IsLocalQueues() {
-        return TFeatureFlags::OptimizationType == EActorSystemOptimizationType::LocalQueues;
-    }
-
 }
diff --git a/ydb/library/actors/core/harmonizer/harmonizer.cpp b/ydb/library/actors/core/harmonizer/harmonizer.cpp
@@ -179,11 +179,11 @@ void THarmonizer::ProcessNeedyState() {
             ProcessingBudget -= 1.0;
             LWPROBE_WITH_DEBUG(HarmonizeOperation, needyPoolIdx, pool.Pool->GetName(), "increase by needs", threadCount + 1, pool.DefaultFullThreadCount, pool.MaxFullThreadCount);
         }
-        if constexpr (NFeatures::IsLocalQueues()) {
+        if (pool.MaxLocalQueueSize) {
             bool needToExpandLocalQueue = ProcessingBudget < 1.0 || threadCount >= pool.MaxFullThreadCount;
             needToExpandLocalQueue &= (bool)pool.BasicPool;
             needToExpandLocalQueue &= (pool.MaxFullThreadCount > 1);
-            needToExpandLocalQueue &= (pool.LocalQueueSize < NFeatures::TLocalQueuesFeatureFlags::MAX_LOCAL_QUEUE_SIZE);
+            needToExpandLocalQueue &= (pool.LocalQueueSize < pool.MaxLocalQueueSize);
             if (needToExpandLocalQueue) {
                 pool.BasicPool->SetLocalQueueSize(++pool.LocalQueueSize);
             }
@@ -258,8 +258,8 @@ void THarmonizer::ProcessHoggishState() {
         if (threadCount == pool.MinFullThreadCount && Shared && SharedInfo.ForeignThreadsAllowed[hoggishPoolIdx] != 0) {
             Shared->SetForeignThreadSlots(hoggishPoolIdx, 0);
         }
-        if (pool.BasicPool && pool.LocalQueueSize > NFeatures::TLocalQueuesFeatureFlags::MIN_LOCAL_QUEUE_SIZE) {
-            pool.LocalQueueSize = std::min<ui16>(NFeatures::TLocalQueuesFeatureFlags::MIN_LOCAL_QUEUE_SIZE, pool.LocalQueueSize / 2);
+        if (pool.BasicPool && pool.LocalQueueSize > pool.MinLocalQueueSize) {
+            pool.LocalQueueSize = std::min<ui16>(pool.MinLocalQueueSize, pool.LocalQueueSize / 2);
             pool.BasicPool->SetLocalQueueSize(pool.LocalQueueSize);
         }
         HARMONIZER_DEBUG_PRINT("poolIdx", hoggishPoolIdx, "threadCount", threadCount, "pool.MinFullThreadCount", pool.MinFullThreadCount, "freeCpu", freeCpu);
@@ -389,6 +389,9 @@ void THarmonizer::AddPool(IExecutorPool* pool, TSelfPingInfo *pingInfo) {
     if (poolInfo.BasicPool) {
         poolInfo.WaitingStats.reset(new TWaitingStats<ui64>());
         poolInfo.MovingWaitingStats.reset(new TWaitingStats<double>());
+        poolInfo.MinLocalQueueSize = poolInfo.BasicPool->GetMinLocalQueueSize();
+        poolInfo.MaxLocalQueueSize = poolInfo.BasicPool->GetMaxLocalQueueSize();
+        poolInfo.LocalQueueSize = poolInfo.MinLocalQueueSize;
     }
     PriorityOrder.clear();
 }
diff --git a/ydb/library/actors/core/harmonizer/pool.cpp b/ydb/library/actors/core/harmonizer/pool.cpp
@@ -12,7 +12,6 @@ namespace NActors {
 LWTRACE_USING(ACTORLIB_PROVIDER);
 
 TPoolInfo::TPoolInfo()
-    : LocalQueueSize(NFeatures::TLocalQueuesFeatureFlags::MIN_LOCAL_QUEUE_SIZE)
 {}
 
 double TPoolInfo::GetCpu(i16 threadIdx) const {
diff --git a/ydb/library/actors/core/harmonizer/pool.h b/ydb/library/actors/core/harmonizer/pool.h
@@ -35,14 +35,17 @@ struct TPoolInfo {
     float MinThreadCount = 0;
     float MaxThreadCount = 0;
 
+    ui16 LocalQueueSize;
+    ui16 MaxLocalQueueSize = 0;
+    ui16 MinLocalQueueSize = 0;
+
     i16 Priority = 0;
     NMonitoring::TDynamicCounters::TCounterPtr AvgPingCounter;
     NMonitoring::TDynamicCounters::TCounterPtr AvgPingCounterWithSmallWindow;
     ui32 MaxAvgPingUs = 0;
     ui64 LastUpdateTs = 0;
     ui64 NotEnoughCpuExecutions = 0;
     ui64 NewNotEnoughCpuExecutions = 0;
-    ui16 LocalQueueSize;
 
     std::atomic<float> SharedCpuQuota = 0;
     std::atomic<i64> LastFlags = 0; // 0 - isNeedy; 1 - isStarved; 2 - isHoggish

Original file line number	Diff line number	Diff line change
`@@ -328,6 +328,13 @@ namespace NKikimr::NAutoConfigInitializer {`
`328`	`328`	`} else {`
`329`	`329`	`executor->SetSpinThreshold(1);`
`330`	`330`	`}`
	`331`	`+`
	`332`	`+ if (config->HasMinLocalQueueSize()) {`
	`333`	`+ executor->SetMinLocalQueueSize(config->GetMinLocalQueueSize());`
	`334`	`+ }`
	`335`	`+ if (config->HasMaxLocalQueueSize()) {`
	`336`	`+ executor->SetMaxLocalQueueSize(config->GetMaxLocalQueueSize());`
	`337`	`+ }`
`331`	`338`	`}`
`332`	`339`	`}`
`333`	`340`
Original file line number	Diff line number	Diff line change
`@@ -86,6 +86,8 @@ message TActorSystemConfig {`
`86`	`86`	`optional int32 MaxAvgPingDeviation = 17;`
`87`	`87`
`88`	`88`	`optional bool HasSharedThread = 18;`
	`89`	`+ optional uint32 MaxLocalQueueSize = 20;`
	`90`	`+ optional uint32 MinLocalQueueSize = 21;`
`89`	`91`	`}`
`90`	`92`
`91`	`93`	`message TScheduler {`
`@@ -125,6 +127,9 @@ message TActorSystemConfig {`
`125`	`127`
`126`	`128`	`optional bool MonitorStuckActors = 15;`
`127`	`129`	`optional EActorSystemProfile ActorSystemProfile = 16;`
	`130`	`+`
	`131`	`+ optional uint32 MinLocalQueueSize = 20;`
	`132`	`+ optional uint32 MaxLocalQueueSize = 21;`
`128`	`133`	`}`
`129`	`134`
`130`	`135`	`message TStaticNameserviceConfig {`
Original file line number	Diff line number	Diff line change
`@@ -179,11 +179,11 @@ void THarmonizer::ProcessNeedyState() {`
`179`	`179`	`ProcessingBudget -= 1.0;`
`180`	`180`	`LWPROBE_WITH_DEBUG(HarmonizeOperation, needyPoolIdx, pool.Pool->GetName(), "increase by needs", threadCount + 1, pool.DefaultFullThreadCount, pool.MaxFullThreadCount);`
`181`	`181`	`}`
`182`		`- if constexpr (NFeatures::IsLocalQueues()) {`
	`182`	`+ if (pool.MaxLocalQueueSize) {`
`183`	`183`	`bool needToExpandLocalQueue = ProcessingBudget < 1.0 \|\| threadCount >= pool.MaxFullThreadCount;`
`184`	`184`	`needToExpandLocalQueue &= (bool)pool.BasicPool;`
`185`	`185`	`needToExpandLocalQueue &= (pool.MaxFullThreadCount > 1);`
`186`		`- needToExpandLocalQueue &= (pool.LocalQueueSize < NFeatures::TLocalQueuesFeatureFlags::MAX_LOCAL_QUEUE_SIZE);`
	`186`	`+ needToExpandLocalQueue &= (pool.LocalQueueSize < pool.MaxLocalQueueSize);`
`187`	`187`	`if (needToExpandLocalQueue) {`
`188`	`188`	`pool.BasicPool->SetLocalQueueSize(++pool.LocalQueueSize);`
`189`	`189`	`}`
`@@ -258,8 +258,8 @@ void THarmonizer::ProcessHoggishState() {`
`258`	`258`	`if (threadCount == pool.MinFullThreadCount && Shared && SharedInfo.ForeignThreadsAllowed[hoggishPoolIdx] != 0) {`
`259`	`259`	`Shared->SetForeignThreadSlots(hoggishPoolIdx, 0);`
`260`	`260`	`}`
`261`		`- if (pool.BasicPool && pool.LocalQueueSize > NFeatures::TLocalQueuesFeatureFlags::MIN_LOCAL_QUEUE_SIZE) {`
`262`		`- pool.LocalQueueSize = std::min<ui16>(NFeatures::TLocalQueuesFeatureFlags::MIN_LOCAL_QUEUE_SIZE, pool.LocalQueueSize / 2);`
	`261`	`+ if (pool.BasicPool && pool.LocalQueueSize > pool.MinLocalQueueSize) {`
	`262`	`+ pool.LocalQueueSize = std::min<ui16>(pool.MinLocalQueueSize, pool.LocalQueueSize / 2);`
`263`	`263`	`pool.BasicPool->SetLocalQueueSize(pool.LocalQueueSize);`
`264`	`264`	`}`
`265`	`265`	`HARMONIZER_DEBUG_PRINT("poolIdx", hoggishPoolIdx, "threadCount", threadCount, "pool.MinFullThreadCount", pool.MinFullThreadCount, "freeCpu", freeCpu);`
`@@ -389,6 +389,9 @@ void THarmonizer::AddPool(IExecutorPool* pool, TSelfPingInfo *pingInfo) {`
`389`	`389`	`if (poolInfo.BasicPool) {`
`390`	`390`	`poolInfo.WaitingStats.reset(new TWaitingStats<ui64>());`
`391`	`391`	`poolInfo.MovingWaitingStats.reset(new TWaitingStats<double>());`
	`392`	`+ poolInfo.MinLocalQueueSize = poolInfo.BasicPool->GetMinLocalQueueSize();`
	`393`	`+ poolInfo.MaxLocalQueueSize = poolInfo.BasicPool->GetMaxLocalQueueSize();`
	`394`	`+ poolInfo.LocalQueueSize = poolInfo.MinLocalQueueSize;`
`392`	`395`	`}`
`393`	`396`	`PriorityOrder.clear();`
`394`	`397`	`}`