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

This reverts commit 4406889. Signed-off-by: Compute-Runtime-Validation <compute-runtime-validation@intel.com>

Author: Compute-Runtime-Validation

shared/source/command_stream/command_stream_receiver.cpp

@@ -673,6 +673,13 @@ 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,6 +352,8 @@ 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,6 +1473,7 @@ inline bool CommandStreamReceiverHw<GfxFamily>::initDirectSubmission() {
  if (directSubmissionController) {
  directSubmissionController->registerDirectSubmission(this);
  }
+ this->startControllingDirectSubmissions();
  if (this->isUpdateTagFromWaitEnabled()) {
  this->overrideDispatchPolicy(DispatchMode::immediateDispatch);
  }
@@ -1485,7 +1486,6 @@ inline bool CommandStreamReceiverHw<GfxFamily>::initDirectSubmission() {
  }
  }
  }
-
  return ret;
 }
 

shared/source/direct_submission/direct_submission_controller.cpp

@@ -66,40 +66,50 @@ void DirectSubmissionController::startThread() {
 }
 
 void DirectSubmissionController::stopThread() {
- {
- std::lock_guard<std::mutex> lock(condVarMutex);
- keepControlling.store(false);
- condVar.notify_one();
- }
+ runControlling.store(false);
+ keepControlling.store(false);
  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 (controller->keepControlling.load()) {
+ while (true) {
+ if (!controller->keepControlling.load()) {
+ return nullptr;
+ }
  std::unique_lock<std::mutex> lock(controller->condVarMutex);
- controller->wait(lock);
- controller->handlePagingFenceRequests(lock);
- controller->sleep(lock);
- controller->handlePagingFenceRequests(lock);
+ controller->handlePagingFenceRequests(lock, true);
+
+ auto isControllerNotified = controller->sleep(lock);
+ if (isControllerNotified) {
+ controller->handlePagingFenceRequests(lock, true);
+ }
  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() {
@@ -111,11 +121,9 @@ void DirectSubmissionController::checkNewSubmissions() {
  std::lock_guard<std::mutex> lock(this->directSubmissionsMutex);
  bool shouldRecalculateTimeout = false;
  std::optional<TaskCountType> bcsTaskCount{};
- for (auto &[csr, state] : directSubmissions) {
- if (!state.isActive) {
- continue;
- }
-
+ for (auto &directSubmission : this->directSubmissions) {
+ auto csr = directSubmission.first;
+ auto &state = directSubmission.second;
  auto isBcs = EngineHelpers::isBcs(csr->getOsContext().getEngineType());
  if (timeoutMode == TimeoutElapsedMode::bcsOnly && !isBcs) {
  continue;
@@ -135,10 +143,8 @@ 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 {
@@ -276,13 +282,13 @@ void DirectSubmissionController::recalculateTimeout() {
 }
 
 void DirectSubmissionController::enqueueWaitForPagingFence(CommandStreamReceiver *csr, uint64_t pagingFenceValue) {
- std::lock_guard lock(condVarMutex);
+ std::lock_guard lock(this->condVarMutex);
  pagingFenceRequests.push({csr, pagingFenceValue});
  condVar.notify_one();
 }
 
 void DirectSubmissionController::drainPagingFenceQueue() {
- std::lock_guard lock(condVarMutex);
+ std::lock_guard lock(this->condVarMutex);
 
  while (!pagingFenceRequests.empty()) {
  auto request = pagingFenceRequests.front();
@@ -291,13 +297,18 @@ void DirectSubmissionController::drainPagingFenceQueue() {
  }
 }
 
-void DirectSubmissionController::handlePagingFenceRequests(std::unique_lock<std::mutex> &lock) {
+void DirectSubmissionController::handlePagingFenceRequests(std::unique_lock<std::mutex> &lock, bool checkForNewSubmissions) {
  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,26 +61,24 @@ 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;
@@ -92,20 +90,15 @@ 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);
- MOCKABLE_VIRTUAL void checkNewSubmissions();
+ 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);
 
@@ -114,7 +107,7 @@ class DirectSubmissionController {
  void updateLastSubmittedThrottle(QueueThrottle throttle);
  size_t getTimeoutParamsMapKey(QueueThrottle throttle, bool acLineStatus);
 
- MOCKABLE_VIRTUAL void handlePagingFenceRequests(std::unique_lock<std::mutex> &lock);
+ void handlePagingFenceRequests(std::unique_lock<std::mutex> &lock, bool checkForNewSubmissions);
  MOCKABLE_VIRTUAL TimeoutElapsedMode timeoutElapsed();
  std::chrono::microseconds getSleepValue() const { return std::chrono::microseconds(this->timeout / this->bcsTimeoutDivisor); }
 
@@ -125,7 +118,7 @@ class DirectSubmissionController {
 
  std::unique_ptr<Thread> directSubmissionControllingThread;
  std::atomic_bool keepControlling = true;
- std::atomic_uint activeSubmissionsCount = 0;
+ std::atomic_bool runControlling = false;
 
  SteadyClock::time_point timeSinceLastCheck{};
  SteadyClock::time_point lastTerminateCpuTimestamp{};
@@ -143,4 +136,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) : csr(commandStreamReceiver), osContext(commandStreamReceiver.getOsContext()), rootDeviceEnvironment(commandStreamReceiver.peekRootDeviceEnvironment()), rootDeviceIndex(commandStreamReceiver.getRootDeviceIndex()) {
+DirectSubmissionInputParams::DirectSubmissionInputParams(const CommandStreamReceiver &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,7 +50,6 @@ class MemoryOperationsHandler;
 
 struct DirectSubmissionInputParams : NonCopyableClass {
  DirectSubmissionInputParams(const CommandStreamReceiver &commandStreamReceiver);
- const CommandStreamReceiver &csr;
  OsContext &osContext;
  const RootDeviceEnvironment &rootDeviceEnvironment;
  MemoryManager *memoryManager = nullptr;
@@ -223,7 +222,6 @@ 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

@@ -10,7 +10,6 @@
 #include "shared/source/command_stream/submissions_aggregator.h"
 #include "shared/source/debug_settings/debug_settings_manager.h"
 #include "shared/source/device/device.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"
@@ -41,7 +40,7 @@ namespace NEO {
 
 template <typename GfxFamily, typename Dispatcher>
 DirectSubmissionHw<GfxFamily, Dispatcher>::DirectSubmissionHw(const DirectSubmissionInputParams &inputParams)
- : ringBuffers(RingBufferUse::initialRingBufferCount), csr(inputParams.csr), osContext(inputParams.osContext), rootDeviceIndex(inputParams.rootDeviceIndex), rootDeviceEnvironment(inputParams.rootDeviceEnvironment) {
+ : ringBuffers(RingBufferUse::initialRingBufferCount), osContext(inputParams.osContext), rootDeviceIndex(inputParams.rootDeviceIndex), rootDeviceEnvironment(inputParams.rootDeviceEnvironment) {
  memoryManager = inputParams.memoryManager;
  globalFenceAllocation = inputParams.globalFenceAllocation;
  hwInfo = inputParams.rootDeviceEnvironment.getHardwareInfo();
@@ -589,9 +588,6 @@ 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

@@ -68,7 +68,7 @@ bool SVMAllocsManager::SvmAllocationCache::insert(size_t size, void *ptr, SvmAll
  return false;
  }
 
- std::unique_lock<std::mutex> lock(this->mtx);
+ std::lock_guard<std::mutex> lock(this->mtx);
  if (svmData->device ? svmData->device->shouldLimitAllocationsReuse() : memoryManager->shouldLimitAllocationsReuse()) {
  return false;
  }
@@ -100,11 +100,8 @@ 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);
- empty = false;
- if (auto usmReuseCleaner = this->memoryManager->peekExecutionEnvironment().unifiedMemoryReuseCleaner.get()) {
- lock.unlock();
- usmReuseCleaner->startThread();
- usmReuseCleaner->notifySvmAllocationsCacheUpdate();
+ if (memoryManager->peekExecutionEnvironment().unifiedMemoryReuseCleaner) {
+ memoryManager->peekExecutionEnvironment().unifiedMemoryReuseCleaner->startThread();
  }
  }
  if (enablePerformanceLogging) {
@@ -114,7 +111,6 @@ bool SVMAllocsManager::SvmAllocationCache::insert(size_t size, void *ptr, SvmAll
  .operationType = CacheOperationType::insert,
  .isSuccess = isSuccess});
  }
-
  return isSuccess;
 }
 
@@ -186,7 +182,6 @@ void *SVMAllocsManager::SvmAllocationCache::get(size_t size, const UnifiedMemory
  svmAllocsManager->reinsertToAllocsForIndirectAccess(*allocationIter->svmData);
  }
  allocations.erase(allocationIter);
- empty = allocations.empty();
  return allocationPtr;
  }
  }
@@ -221,7 +216,6 @@ void SVMAllocsManager::SvmAllocationCache::trim() {
  svmAllocsManager->freeSVMAllocImpl(cachedAllocationInfo.allocation, FreePolicyType::blocking, cachedAllocationInfo.svmData);
  }
  this->allocations.clear();
- empty = true;
 }
 
 void SVMAllocsManager::SvmAllocationCache::cleanup() {
@@ -306,7 +300,6 @@ 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

@@ -221,7 +221,6 @@ 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);
@@ -235,7 +234,6 @@ class SVMAllocsManager {
  MemoryManager *memoryManager = nullptr;
  bool enablePerformanceLogging = false;
  bool requireUpdatingAllocsForIndirectAccess = false;
- std::atomic_bool empty = true;
  };
 
  enum class FreePolicyType : uint32_t {