fix: use condition variables instead of busy waits in worker threads

Resolves: NEO-16085, GSD-11678, HSD-14025819208

Signed-off-by: Igor Venevtsev <igor.venevtsev@intel.com>

Author: ivvenevt

shared/source/command_stream/command_stream_receiver.cpp

@@ -673,13 +673,6 @@ void CommandStreamReceiver::downloadAllocation(GraphicsAllocation &gfxAllocation
     }
 }
 
-void CommandStreamReceiver::startControllingDirectSubmissions() {
-    auto controller = this->executionEnvironment.directSubmissionController.get();
-    if (controller) {
-        controller->startControlling();
-    }
-}
-
 bool CommandStreamReceiver::enqueueWaitForPagingFence(uint64_t pagingFenceValue) {
     auto controller = this->executionEnvironment.directSubmissionController.get();
     if (this->isAnyDirectSubmissionEnabled() && controller) {

shared/source/command_stream/command_stream_receiver.h

@@ -352,8 +352,6 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
 
     uint32_t getRootDeviceIndex() const { return rootDeviceIndex; }
 
-    MOCKABLE_VIRTUAL void startControllingDirectSubmissions();
-
     MOCKABLE_VIRTUAL bool isAnyDirectSubmissionEnabled() const {
         return this->isDirectSubmissionEnabled() || isBlitterDirectSubmissionEnabled();
     }

shared/source/command_stream/command_stream_receiver_hw_base.inl

@@ -1304,7 +1304,7 @@ SubmissionStatus CommandStreamReceiverHw<GfxFamily>::flushSmallTask(LinearStream
     this->latestSentTaskCount = taskCount + 1;
     auto submissionStatus = flushHandler(batchBuffer, getResidencyAllocations());
     if (submissionStatus == SubmissionStatus::success) {
-        taskCount++;
+        ++taskCount;
     }
     return submissionStatus;
 }
@@ -1473,7 +1473,6 @@ inline bool CommandStreamReceiverHw<GfxFamily>::initDirectSubmission() {
                 if (directSubmissionController) {
                     directSubmissionController->registerDirectSubmission(this);
                 }
-                this->startControllingDirectSubmissions();
                 if (this->isUpdateTagFromWaitEnabled()) {
                     this->overrideDispatchPolicy(DispatchMode::immediateDispatch);
                 }
@@ -1486,6 +1485,7 @@ inline bool CommandStreamReceiverHw<GfxFamily>::initDirectSubmission() {
             }
         }
     }
+
     return ret;
 }
 

shared/source/direct_submission/direct_submission_controller.cpp

@@ -66,50 +66,40 @@ void DirectSubmissionController::startThread() {
 }
 
 void DirectSubmissionController::stopThread() {
-    runControlling.store(false);
-    keepControlling.store(false);
+    {
+        std::lock_guard<std::mutex> lock(condVarMutex);
+        keepControlling.store(false);
+        condVar.notify_one();
+    }
     if (directSubmissionControllingThread) {
         directSubmissionControllingThread->join();
         directSubmissionControllingThread.reset();
     }
 }
 
-void DirectSubmissionController::startControlling() {
-    this->runControlling.store(true);
-}
-
 void *DirectSubmissionController::controlDirectSubmissionsState(void *self) {
     auto controller = reinterpret_cast<DirectSubmissionController *>(self);
 
-    while (!controller->runControlling.load()) {
-        if (!controller->keepControlling.load()) {
-            return nullptr;
-        }
-        std::unique_lock<std::mutex> lock(controller->condVarMutex);
-        controller->handlePagingFenceRequests(lock, false);
-
-        auto isControllerNotified = controller->sleep(lock);
-        if (isControllerNotified) {
-            controller->handlePagingFenceRequests(lock, false);
-        }
-    }
-
     controller->timeSinceLastCheck = controller->getCpuTimestamp();
     controller->lastHangCheckTime = std::chrono::high_resolution_clock::now();
-    while (true) {
-        if (!controller->keepControlling.load()) {
-            return nullptr;
-        }
-        std::unique_lock<std::mutex> lock(controller->condVarMutex);
-        controller->handlePagingFenceRequests(lock, true);
 
-        auto isControllerNotified = controller->sleep(lock);
-        if (isControllerNotified) {
-            controller->handlePagingFenceRequests(lock, true);
-        }
+    while (controller->keepControlling.load()) {
+        std::unique_lock<std::mutex> lock(controller->condVarMutex);
+        controller->wait(lock);
+        controller->handlePagingFenceRequests(lock);
+        controller->sleep(lock);
+        controller->handlePagingFenceRequests(lock);
         lock.unlock();
         controller->checkNewSubmissions();
     }
+
+    return nullptr;
+}
+
+void DirectSubmissionController::notifyNewSubmission(const CommandStreamReceiver *csr) {
+    ++activeSubmissionsCount;
+    directSubmissions[const_cast<CommandStreamReceiver *>(csr)].isActive = true;
+    condVar.notify_one();
 }
 
 void DirectSubmissionController::checkNewSubmissions() {
@@ -121,9 +111,11 @@ void DirectSubmissionController::checkNewSubmissions() {
     std::lock_guard<std::mutex> lock(this->directSubmissionsMutex);
     bool shouldRecalculateTimeout = false;
     std::optional<TaskCountType> bcsTaskCount{};
-    for (auto &directSubmission : this->directSubmissions) {
-        auto csr = directSubmission.first;
-        auto &state = directSubmission.second;
+    for (auto &[csr, state] : directSubmissions) {
+        if (!state.isActive) {
+            continue;
+        }
+
         auto isBcs = EngineHelpers::isBcs(csr->getOsContext().getEngineType());
         if (timeoutMode == TimeoutElapsedMode::bcsOnly && !isBcs) {
             continue;
@@ -143,8 +135,10 @@ void DirectSubmissionController::checkNewSubmissions() {
             auto lock = csr->obtainUniqueOwnership();
             if (!isCsrIdleDetectionEnabled || (isDirectSubmissionIdle(csr, lock) && isCopyEngineIdle)) {
                 csr->stopDirectSubmission(false, false);
+                state.isActive = false;
                 state.isStopped = true;
                 shouldRecalculateTimeout = true;
+                --activeSubmissionsCount;
             }
             state.taskCount = csr->peekTaskCount();
         } else {
@@ -282,13 +276,13 @@ void DirectSubmissionController::recalculateTimeout() {
 }
 
 void DirectSubmissionController::enqueueWaitForPagingFence(CommandStreamReceiver *csr, uint64_t pagingFenceValue) {
-    std::lock_guard lock(this->condVarMutex);
+    std::lock_guard lock(condVarMutex);
     pagingFenceRequests.push({csr, pagingFenceValue});
     condVar.notify_one();
 }
 
 void DirectSubmissionController::drainPagingFenceQueue() {
-    std::lock_guard lock(this->condVarMutex);
+    std::lock_guard lock(condVarMutex);
 
     while (!pagingFenceRequests.empty()) {
         auto request = pagingFenceRequests.front();
@@ -297,18 +291,13 @@ void DirectSubmissionController::drainPagingFenceQueue() {
     }
 }
 
-void DirectSubmissionController::handlePagingFenceRequests(std::unique_lock<std::mutex> &lock, bool checkForNewSubmissions) {
+void DirectSubmissionController::handlePagingFenceRequests(std::unique_lock<std::mutex> &lock) {
     UNRECOVERABLE_IF(!lock.owns_lock())
     while (!pagingFenceRequests.empty()) {
         auto request = pagingFenceRequests.front();
         pagingFenceRequests.pop();
         lock.unlock();
-
         request.csr->unblockPagingFenceSemaphore(request.pagingFenceValue);
-        if (checkForNewSubmissions) {
-            checkNewSubmissions();
-        }
-
         lock.lock();
     }
 }

shared/source/direct_submission/direct_submission_controller.h

@@ -61,24 +61,26 @@ class DirectSubmissionController {
     void unregisterDirectSubmission(CommandStreamReceiver *csr);
 
     void startThread();
-    void startControlling();
     void stopThread();
 
     static bool isSupported();
 
     void enqueueWaitForPagingFence(CommandStreamReceiver *csr, uint64_t pagingFenceValue);
     void drainPagingFenceQueue();
+    void notifyNewSubmission(const CommandStreamReceiver *csr);
 
   protected:
     struct DirectSubmissionState {
         DirectSubmissionState(DirectSubmissionState &&other) noexcept {
+            isActive = other.isActive.load();
             isStopped = other.isStopped.load();
             taskCount = other.taskCount.load();
         }
         DirectSubmissionState &operator=(const DirectSubmissionState &other) {
             if (this == &other) {
                 return *this;
             }
+            this->isActive = other.isActive.load();
             this->isStopped = other.isStopped.load();
             this->taskCount = other.taskCount.load();
             return *this;
@@ -90,15 +92,20 @@ class DirectSubmissionController {
         DirectSubmissionState(const DirectSubmissionState &other) = delete;
         DirectSubmissionState &operator=(DirectSubmissionState &&other) = delete;
 
+        std::atomic_bool isActive{false};
         std::atomic_bool isStopped{true};
         std::atomic<TaskCountType> taskCount{0};
     };
 
     static void *controlDirectSubmissionsState(void *self);
-    void checkNewSubmissions();
+    MOCKABLE_VIRTUAL void checkNewSubmissions();
     bool isDirectSubmissionIdle(CommandStreamReceiver *csr, std::unique_lock<std::recursive_mutex> &csrLock);
     bool isCopyEngineOnDeviceIdle(uint32_t rootDeviceIndex, std::optional<TaskCountType> &bcsTaskCount);
     MOCKABLE_VIRTUAL bool sleep(std::unique_lock<std::mutex> &lock);
+    bool waitPredicate() { return !keepControlling || !pagingFenceRequests.empty() || activeSubmissionsCount; }
+    MOCKABLE_VIRTUAL void wait(std::unique_lock<std::mutex> &lock) {
+        condVar.wait(lock, [&]() { return waitPredicate(); });
+    }
     MOCKABLE_VIRTUAL SteadyClock::time_point getCpuTimestamp();
     MOCKABLE_VIRTUAL void overrideDirectSubmissionTimeouts(const ProductHelper &productHelper);
 
@@ -107,7 +114,7 @@ class DirectSubmissionController {
     void updateLastSubmittedThrottle(QueueThrottle throttle);
     size_t getTimeoutParamsMapKey(QueueThrottle throttle, bool acLineStatus);
 
-    void handlePagingFenceRequests(std::unique_lock<std::mutex> &lock, bool checkForNewSubmissions);
+    MOCKABLE_VIRTUAL void handlePagingFenceRequests(std::unique_lock<std::mutex> &lock);
     MOCKABLE_VIRTUAL TimeoutElapsedMode timeoutElapsed();
     std::chrono::microseconds getSleepValue() const { return std::chrono::microseconds(this->timeout / this->bcsTimeoutDivisor); }
 
@@ -118,7 +125,7 @@ class DirectSubmissionController {
 
     std::unique_ptr<Thread> directSubmissionControllingThread;
     std::atomic_bool keepControlling = true;
-    std::atomic_bool runControlling = false;
+    std::atomic_uint activeSubmissionsCount = 0;
 
     SteadyClock::time_point timeSinceLastCheck{};
     SteadyClock::time_point lastTerminateCpuTimestamp{};
@@ -136,4 +143,4 @@ class DirectSubmissionController {
 
     std::queue<WaitForPagingFenceRequest> pagingFenceRequests;
 };
-} // namespace NEO
+} // namespace NEO

shared/source/direct_submission/direct_submission_hw.cpp

@@ -10,7 +10,7 @@
 #include "shared/source/command_stream/command_stream_receiver.h"
 
 namespace NEO {
-DirectSubmissionInputParams::DirectSubmissionInputParams(const CommandStreamReceiver &commandStreamReceiver) : osContext(commandStreamReceiver.getOsContext()), rootDeviceEnvironment(commandStreamReceiver.peekRootDeviceEnvironment()), rootDeviceIndex(commandStreamReceiver.getRootDeviceIndex()) {
+DirectSubmissionInputParams::DirectSubmissionInputParams(const CommandStreamReceiver &commandStreamReceiver) : csr(commandStreamReceiver), osContext(commandStreamReceiver.getOsContext()), rootDeviceEnvironment(commandStreamReceiver.peekRootDeviceEnvironment()), rootDeviceIndex(commandStreamReceiver.getRootDeviceIndex()) {
     memoryManager = commandStreamReceiver.getMemoryManager();
     globalFenceAllocation = commandStreamReceiver.getGlobalFenceAllocation();
     workPartitionAllocation = commandStreamReceiver.getWorkPartitionAllocation();

shared/source/direct_submission/direct_submission_hw.h

@@ -50,6 +50,7 @@ class MemoryOperationsHandler;
 
 struct DirectSubmissionInputParams : NonCopyableClass {
     DirectSubmissionInputParams(const CommandStreamReceiver &commandStreamReceiver);
+    const CommandStreamReceiver &csr;
     OsContext &osContext;
     const RootDeviceEnvironment &rootDeviceEnvironment;
     MemoryManager *memoryManager = nullptr;
@@ -222,6 +223,7 @@ class DirectSubmissionHw {
     uint64_t gpuVaForPagingFenceSemaphore = 0u;
     uint64_t relaxedOrderingQueueSizeLimitValueVa = 0;
 
+    const CommandStreamReceiver &csr;
     OsContext &osContext;
     const uint32_t rootDeviceIndex;
     MemoryManager *memoryManager = nullptr;

shared/source/direct_submission/direct_submission_hw.inl

@@ -8,6 +8,7 @@
 #include "shared/source/command_container/command_encoder.h"
 #include "shared/source/command_stream/submissions_aggregator.h"
 #include "shared/source/debug_settings/debug_settings_manager.h"
+#include "shared/source/direct_submission/direct_submission_controller.h"
 #include "shared/source/direct_submission/direct_submission_hw.h"
 #include "shared/source/direct_submission/relaxed_ordering_helper.h"
 #include "shared/source/execution_environment/execution_environment.h"
@@ -33,7 +34,7 @@ namespace NEO {
 
 template <typename GfxFamily, typename Dispatcher>
 DirectSubmissionHw<GfxFamily, Dispatcher>::DirectSubmissionHw(const DirectSubmissionInputParams &inputParams)
-    : ringBuffers(RingBufferUse::initialRingBufferCount), osContext(inputParams.osContext), rootDeviceIndex(inputParams.rootDeviceIndex), rootDeviceEnvironment(inputParams.rootDeviceEnvironment) {
+    : ringBuffers(RingBufferUse::initialRingBufferCount), csr(inputParams.csr), osContext(inputParams.osContext), rootDeviceIndex(inputParams.rootDeviceIndex), rootDeviceEnvironment(inputParams.rootDeviceEnvironment) {
     memoryManager = inputParams.memoryManager;
     globalFenceAllocation = inputParams.globalFenceAllocation;
     hwInfo = inputParams.rootDeviceEnvironment.getHardwareInfo();
@@ -581,6 +582,9 @@ template <typename GfxFamily, typename Dispatcher>
 bool DirectSubmissionHw<GfxFamily, Dispatcher>::submitCommandBufferToGpu(bool needStart, uint64_t gpuAddress, size_t size, bool needWait, const ResidencyContainer *allocationsForResidency) {
     if (needStart) {
         this->ringStart = this->submit(gpuAddress, size, allocationsForResidency);
+        if (auto controller = rootDeviceEnvironment.executionEnvironment.directSubmissionController.get()) {
+            controller->notifyNewSubmission(&csr);
+        }
         return this->ringStart;
     } else {
         if (needWait) {

shared/source/memory_manager/unified_memory_manager.cpp

@@ -69,7 +69,7 @@ bool SVMAllocsManager::SvmAllocationCache::insert(size_t size, void *ptr, SvmAll
         return false;
     }
 
-    std::lock_guard<std::mutex> lock(this->mtx);
+    std::unique_lock<std::mutex> lock(this->mtx);
     if (svmData->device ? svmData->device->shouldLimitAllocationsReuse() : memoryManager->shouldLimitAllocationsReuse()) {
         return false;
     }
@@ -101,8 +101,11 @@ bool SVMAllocsManager::SvmAllocationCache::insert(size_t size, void *ptr, SvmAll
         }
         svmData->isSavedForReuse = true;
         allocations.emplace(std::lower_bound(allocations.begin(), allocations.end(), size), size, ptr, svmData, waitForCompletion);
-        if (memoryManager->peekExecutionEnvironment().unifiedMemoryReuseCleaner) {
-            memoryManager->peekExecutionEnvironment().unifiedMemoryReuseCleaner->startThread();
+        empty = false;
+        if (auto usmReuseCleaner = this->memoryManager->peekExecutionEnvironment().unifiedMemoryReuseCleaner.get()) {
+            lock.unlock();
+            usmReuseCleaner->startThread();
+            usmReuseCleaner->notifySvmAllocationsCacheUpdate();
         }
     }
     if (enablePerformanceLogging) {
@@ -112,6 +115,7 @@ bool SVMAllocsManager::SvmAllocationCache::insert(size_t size, void *ptr, SvmAll
                            .operationType = CacheOperationType::insert,
                            .isSuccess = isSuccess});
     }
+
     return isSuccess;
 }
 
@@ -183,6 +187,7 @@ void *SVMAllocsManager::SvmAllocationCache::get(size_t size, const UnifiedMemory
                 svmAllocsManager->reinsertToAllocsForIndirectAccess(*allocationIter->svmData);
             }
             allocations.erase(allocationIter);
+            empty = allocations.empty();
             return allocationPtr;
         }
     }
@@ -217,6 +222,7 @@ void SVMAllocsManager::SvmAllocationCache::trim() {
         svmAllocsManager->freeSVMAllocImpl(cachedAllocationInfo.allocation, FreePolicyType::blocking, cachedAllocationInfo.svmData);
     }
     this->allocations.clear();
+    empty = true;
 }
 
 void SVMAllocsManager::SvmAllocationCache::cleanup() {
@@ -301,6 +307,7 @@ void SVMAllocsManager::SvmAllocationCache::trimOldAllocs(std::chrono::high_resol
     if (trimAll) {
         std::erase_if(allocations, SvmCacheAllocationInfo::isMarkedForDelete);
     }
+    empty = allocations.empty();
 }
 
 SvmAllocationData *SVMAllocsManager::MapBasedAllocationTracker::get(const void *ptr) {

shared/source/memory_manager/unified_memory_manager.h

@@ -203,6 +203,7 @@ class SVMAllocsManager {
         static bool allocUtilizationAllows(size_t requestedSize, size_t reuseCandidateSize);
         static bool alignmentAllows(void *ptr, size_t alignment);
         bool isInUse(SvmCacheAllocationInfo &cacheAllocInfo);
+        bool isEmpty() { return empty; };
         void *get(size_t size, const UnifiedMemoryProperties &unifiedMemoryProperties);
         void trim();
         void trimOldAllocs(std::chrono::high_resolution_clock::time_point trimTimePoint, bool trimAll);
@@ -216,6 +217,7 @@ class SVMAllocsManager {
         MemoryManager *memoryManager = nullptr;
         bool enablePerformanceLogging = false;
         bool requireUpdatingAllocsForIndirectAccess = false;
+        std::atomic_bool empty = true;
     };
 
     enum class FreePolicyType : uint32_t {