Improve ring queue perfomance

Author: kruall

ydb/library/actors/core/executor_pool_basic.cpp

@@ -264,6 +264,63 @@ namespace NActors {
         return nullptr;
     }
 
+    TMailbox* TBasicExecutorPool::GetReadyActivationRingQueue(ui64 revolvingCounter) {
+        if (StopFlag.load(std::memory_order_acquire)) {
+            return nullptr;
+        }
+        
+        TWorkerId workerId = TlsThreadContext->WorkerId();
+        EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "");
+        NHPTimer::STime hpnow = GetCycleCountFast();
+        TInternalActorTypeGuard<EInternalActorSystemActivity::ACTOR_SYSTEM_GET_ACTIVATION, false> activityGuard(hpnow);
+
+        Y_DEBUG_ABORT_UNLESS(workerId < MaxFullThreadCount);
+
+        Threads[workerId].UnsetWork();
+        if (Harmonizer) {
+            EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "try to harmonize");
+            LWPROBE(TryToHarmonize, PoolId, PoolName);
+            Harmonizer->Harmonize(hpnow);
+            EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "harmonize done");
+        }
+
+        do {
+            {
+                TInternalActorTypeGuard<EInternalActorSystemActivity::ACTOR_SYSTEM_GET_ACTIVATION_FROM_QUEUE, false> activityGuard;
+                if (const ui32 activation = std::visit([&revolvingCounter](auto &x) {return x.Pop(++revolvingCounter);}, Activations)) {
+                    EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "activation found");
+                    Threads[workerId].SetWork();
+                    AtomicDecrement(Semaphore);
+                    return MailboxTable->Get(activation);
+                }
+            }
+
+            TAtomic semaphoreRaw = AtomicGet(Semaphore);
+            TSemaphore semaphore = TSemaphore::GetSemaphore(semaphoreRaw);
+            if (!semaphore.OldSemaphore || workerId >= 0 && semaphore.CurrentSleepThreadCount < 0) {
+                EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "semaphore.OldSemaphore == 0 or workerId >= 0 && semaphore.CurrentSleepThreadCount < 0");
+                if (!TlsThreadContext->ExecutionContext.IsNeededToWaitNextActivation) {
+                    EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "wctx.ExecutionContext.IsNeededToWaitNextActivation == false");
+                    return nullptr;
+                }
+
+                bool needToWait = false;
+                bool needToBlock = false;
+                AskToGoToSleep(&needToWait, &needToBlock);
+                if (needToWait) {
+                    EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "go to sleep");
+                    if (Threads[workerId].Wait(SpinThresholdCycles, &StopFlag)) {
+                        EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "sleep interrupted");
+                        return nullptr;
+                    }
+                }
+            }
+            SpinLockPause();
+        } while (!StopFlag.load(std::memory_order_acquire));
+
+        return nullptr;
+    }
+
     TMailbox* TBasicExecutorPool::GetReadyActivationLocalQueue(ui64 revolvingCounter) {
         TWorkerId workerId = TlsThreadContext->WorkerId();
         Y_DEBUG_ABORT_UNLESS(workerId < static_cast<i32>(MaxFullThreadCount));
@@ -278,13 +335,19 @@ namespace NActors {
             TlsThreadContext->LocalQueueContext.LocalQueueSize = LocalQueueSize.load(std::memory_order_relaxed);
         }
         EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "local queue done; moving to common");
+        if (TlsThreadContext->UseRingQueue()) {
+            return GetReadyActivationRingQueue(revolvingCounter);
+        }
         return GetReadyActivationCommon(revolvingCounter);
     }
 
     TMailbox* TBasicExecutorPool::GetReadyActivation(ui64 revolvingCounter) {
         if (MaxLocalQueueSize) {
             EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "local queue");
             return GetReadyActivationLocalQueue(revolvingCounter);
+        } else if (TlsThreadContext->UseRingQueue()) {
+            EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "ring queue");
+            return GetReadyActivationRingQueue(revolvingCounter);
         } else {
             EXECUTOR_POOL_BASIC_DEBUG(EDebugLevel::Activation, "");
             return GetReadyActivationCommon(revolvingCounter);

ydb/library/actors/core/executor_pool_basic.h

@@ -234,6 +234,7 @@ namespace NActors {
         TMailbox* GetReadyActivation(ui64 revolvingReadCounter) override;
         TMailbox* GetReadyActivationCommon(ui64 revolvingReadCounter);
         TMailbox* GetReadyActivationShared(ui64 revolvingReadCounter);
+        TMailbox* GetReadyActivationRingQueue(ui64 revolvingReadCounter);
         TMailbox* GetReadyActivationLocalQueue(ui64 revolvingReadCounter);
 
         void Schedule(TInstant deadline, TAutoPtr<IEventHandle> ev, ISchedulerCookie* cookie, TWorkerId workerId) override;