merge 24-3: add ring queue config

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -303,6 +303,7 @@ void AddExecutorPool(
         TBasicExecutorPoolConfig basic;
         basic.PoolId = poolId;
         basic.PoolName = poolConfig.GetName();
+        basic.UseRingQueue = systemConfig.HasUseRingQueue() && systemConfig.GetUseRingQueue();
         if (poolConfig.HasMaxAvgPingDeviation()) {
             auto poolGroup = counters->GetSubgroup("execpool", basic.PoolName);
             auto &poolInfo = cpuManager.PingInfoByPool[poolId];

ydb/core/protos/config.proto

@@ -118,6 +118,7 @@ message TActorSystemConfig {
     optional ENodeType NodeType = 14 [default = COMPUTE];
     optional uint32 ForceIOPoolThreads = 17;
     optional bool UseSharedThreads = 18;
+    optional bool UseRingQueue = 19;
 
     optional bool MonitorStuckActors = 15;
     optional EActorSystemProfile ActorSystemProfile = 16;

ydb/library/actors/core/config.h

@@ -31,6 +31,7 @@ namespace NActors {
         i16 SoftProcessingDurationTs = 0;
         EASProfile ActorSystemProfile = EASProfile::Default;
         bool HasSharedThread = false;
+        bool UseRingQueue = false;
     };
 
     struct TSharedExecutorPoolConfig {
@@ -47,6 +48,7 @@ namespace NActors {
         TString PoolName;
         ui32 Threads = 1;
         TCpuMask Affinity; // Executor thread affinity
+        bool UseRingQueue = false;
     };
 
     struct TSelfPingInfo {

ydb/library/actors/core/executor_pool_base.cpp

@@ -65,17 +65,20 @@ namespace NActors {
     }
 #endif
 
-    TExecutorPoolBase::TExecutorPoolBase(ui32 poolId, ui32 threads, TAffinity* affinity)
+    TExecutorPoolBase::TExecutorPoolBase(ui32 poolId, ui32 threads, TAffinity* affinity, bool useRingQueue)
         : TExecutorPoolBaseMailboxed(poolId)
         , PoolThreads(threads)
         , ThreadsAffinity(affinity)
-#ifdef RING_ACTIVATION_QUEUE
-        , Activations(threads == 1)
-#endif
-    {}
+    {
+        if (useRingQueue) {
+            Activations.emplace<TRingActivationQueue>(threads == 1);
+        } else {
+            Activations.emplace<TUnorderedCacheActivationQueue>();
+        }
+    }
 
     TExecutorPoolBase::~TExecutorPoolBase() {
-        while (Activations.Pop(0))
+        while (std::visit([](auto &x){return x.Pop(0);}, Activations))
             ;
     }
 

ydb/library/actors/core/executor_pool_base.h

@@ -47,21 +47,16 @@ namespace NActors {
 
     class TExecutorPoolBase: public TExecutorPoolBaseMailboxed {
     protected:
-
-#ifdef RING_ACTIVATION_QUEUE
-        using TActivationQueue = TRingActivationQueue;
-#else
-        using TActivationQueue = TUnorderedCache<ui32, 512, 4>;
-#endif
+        using TUnorderedCacheActivationQueue = TUnorderedCache<ui32, 512, 4>;
 
         const i16 PoolThreads;
         TIntrusivePtr<TAffinity> ThreadsAffinity;
         TAtomic Semaphore = 0;
-        TActivationQueue Activations;
+        std::variant<TUnorderedCacheActivationQueue, TRingActivationQueue> Activations;
         TAtomic ActivationsRevolvingCounter = 0;
         std::atomic_bool StopFlag = false;
     public:
-        TExecutorPoolBase(ui32 poolId, ui32 threads, TAffinity* affinity);
+        TExecutorPoolBase(ui32 poolId, ui32 threads, TAffinity* affinity, bool useRingQueue);
         ~TExecutorPoolBase();
         void ScheduleActivation(ui32 activation) override;
         void SpecificScheduleActivation(ui32 activation) override;

ydb/library/actors/core/executor_pool_basic.cpp

@@ -82,7 +82,7 @@ namespace NActors {
     }
 
     TBasicExecutorPool::TBasicExecutorPool(const TBasicExecutorPoolConfig& cfg, IHarmonizer *harmonizer, TExecutorPoolJail *jail)
-        : TExecutorPoolBase(cfg.PoolId, cfg.Threads, new TAffinity(cfg.Affinity))
+        : TExecutorPoolBase(cfg.PoolId, cfg.Threads, new TAffinity(cfg.Affinity), cfg.UseRingQueue)
         , DefaultSpinThresholdCycles(cfg.SpinThreshold * NHPTimer::GetCyclesPerSecond() * 0.000001) // convert microseconds to cycles
         , SpinThresholdCycles(DefaultSpinThresholdCycles)
         , SpinThresholdCyclesPerThread(new NThreading::TPadded<std::atomic<ui64>>[cfg.Threads])
@@ -235,7 +235,7 @@ namespace NActors {
                 }
             } else {
                 TInternalActorTypeGuard<EInternalActorSystemActivity::ACTOR_SYSTEM_GET_ACTIVATION_FROM_QUEUE, false> activityGuard;
-                if (const ui32 activation = Activations.Pop(++revolvingCounter)) {
+                if (const ui32 activation = std::visit([&revolvingCounter](auto &x) {return x.Pop(++revolvingCounter);}, Activations)) {
                     if (workerId >= 0) {
                         Threads[workerId].SetWork();
                     } else {
@@ -308,8 +308,9 @@ namespace NActors {
 
     void TBasicExecutorPool::ScheduleActivationExCommon(ui32 activation, ui64 revolvingCounter, TAtomic x) {
         TSemaphore semaphore = TSemaphore::GetSemaphore(x);
-
-        Activations.Push(activation, revolvingCounter);
+        std::visit([activation, revolvingCounter](auto &x) {
+            x.Push(activation, revolvingCounter);
+        }, Activations);
         bool needToWakeUp = false;
         bool needToChangeOldSemaphore = true;
 

ydb/library/actors/core/executor_pool_io.cpp

@@ -6,8 +6,8 @@
 #include <ydb/library/actors/util/datetime.h>
 
 namespace NActors {
-    TIOExecutorPool::TIOExecutorPool(ui32 poolId, ui32 threads, const TString& poolName, TAffinity* affinity)
-        : TExecutorPoolBase(poolId, threads, affinity)
+    TIOExecutorPool::TIOExecutorPool(ui32 poolId, ui32 threads, const TString& poolName, TAffinity* affinity, bool useRingQueue)
+        : TExecutorPoolBase(poolId, threads, affinity, useRingQueue)
         , Threads(new TExecutorThreadCtx[threads])
         , PoolName(poolName)
     {}
@@ -17,7 +17,8 @@ namespace NActors {
             cfg.PoolId,
             cfg.Threads,
             cfg.PoolName,
-            new TAffinity(cfg.Affinity)
+            new TAffinity(cfg.Affinity),
+            cfg.UseRingQueue
         )
     {
         Harmonizer = harmonizer;
@@ -53,7 +54,7 @@ namespace NActors {
         }
 
         while (!StopFlag.load(std::memory_order_acquire)) {
-            if (const ui32 activation = Activations.Pop(++revolvingCounter)) {
+            if (const ui32 activation = std::visit([&revolvingCounter](auto &x){return x.Pop(++revolvingCounter);}, Activations)) {
                 return activation;
             }
             SpinLockPause();
@@ -86,7 +87,9 @@ namespace NActors {
     }
 
     void TIOExecutorPool::ScheduleActivationEx(ui32 activation, ui64 revolvingWriteCounter) {
-        Activations.Push(activation, revolvingWriteCounter);
+        std::visit([activation, revolvingWriteCounter](auto &x) {
+            x.Push(activation, revolvingWriteCounter);
+        }, Activations);
         const TAtomic x = AtomicIncrement(Semaphore);
         if (x <= 0) {
             for (;; ++revolvingWriteCounter) {

ydb/library/actors/core/executor_pool_io.h

@@ -26,7 +26,7 @@ namespace NActors {
         const TString PoolName;
         const ui32 ActorSystemIndex = NActors::TActorTypeOperator::GetActorSystemIndex();
     public:
-        TIOExecutorPool(ui32 poolId, ui32 threads, const TString& poolName = "", TAffinity* affinity = nullptr);
+        TIOExecutorPool(ui32 poolId, ui32 threads, const TString& poolName = "", TAffinity* affinity = nullptr, bool useRingQueue = false);
         explicit TIOExecutorPool(const TIOExecutorPoolConfig& cfg, IHarmonizer *harmonizer = nullptr);
         ~TIOExecutorPool();