Change activation queue to variant of queues (#8078)

Author: kruall

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), false)
         , 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

@@ -7,7 +7,7 @@
 
 namespace NActors {
     TIOExecutorPool::TIOExecutorPool(ui32 poolId, ui32 threads, const TString& poolName, TAffinity* affinity)
-        : TExecutorPoolBase(poolId, threads, affinity)
+        : TExecutorPoolBase(poolId, threads, affinity, false)
         , Threads(new TExecutorThreadCtx[threads])
         , PoolName(poolName)
     {}
@@ -53,7 +53,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 +86,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) {