From 446c2c027e100af67bccdc20ae997bb05e782db8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Henrik=20Rydg=C3=A5rd?= Date: Mon, 30 Nov 2020 16:46:52 +0100 Subject: [PATCH] Get rid of the PrioritizedWorkQueue. Instead just queue tasks on the ThreadManager. --- CMakeLists.txt | 2 - Common/Common.vcxproj | 6 +- Common/Common.vcxproj.filters | 9 +- Common/Thread/Event.h | 27 +++++ Common/Thread/ParallelLoop.cpp | 2 +- Common/Thread/ParallelLoop.h | 9 +- Common/Thread/PrioritizedWorkQueue.cpp | 140 ------------------------ Common/Thread/PrioritizedWorkQueue.h | 65 ----------- Common/Thread/Promise.h | 2 +- Common/Thread/ThreadManager.cpp | 4 +- Common/Thread/ThreadManager.h | 28 ++++- UI/GameInfoCache.cpp | 80 +++++--------- UI/GameInfoCache.h | 8 +- UWP/CommonUWP/CommonUWP.vcxproj | 4 +- UWP/CommonUWP/CommonUWP.vcxproj.filters | 4 +- android/jni/Android.mk | 4 + libretro/Makefile.common | 1 - 17 files changed, 102 insertions(+), 293 deletions(-) create mode 100644 Common/Thread/Event.h delete mode 100644 Common/Thread/PrioritizedWorkQueue.cpp delete mode 100644 Common/Thread/PrioritizedWorkQueue.h diff --git a/CMakeLists.txt b/CMakeLists.txt index f32da2d8b2..8e73c3abf8 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -611,8 +611,6 @@ add_library(Common STATIC Common/Thread/Channel.h Common/Thread/ParallelLoop.cpp Common/Thread/ParallelLoop.h - Common/Thread/PrioritizedWorkQueue.cpp - Common/Thread/PrioritizedWorkQueue.h Common/Thread/Promise.h Common/Thread/ThreadUtil.cpp Common/Thread/ThreadUtil.h diff --git a/Common/Common.vcxproj b/Common/Common.vcxproj index b368ea2155..e4264b4f20 100644 --- a/Common/Common.vcxproj +++ b/Common/Common.vcxproj @@ -528,8 +528,9 @@ + + - @@ -968,7 +969,6 @@ - @@ -999,4 +999,4 @@ - \ No newline at end of file + diff --git a/Common/Common.vcxproj.filters b/Common/Common.vcxproj.filters index 245dd70e2b..58ca55dbbf 100644 --- a/Common/Common.vcxproj.filters +++ b/Common/Common.vcxproj.filters @@ -90,9 +90,6 @@ ext\libpng17 - - Thread - Thread @@ -407,6 +404,9 @@ Thread + + Thread + @@ -502,9 +502,6 @@ ext\libpng17 - - Thread - Thread diff --git a/Common/Thread/Event.h b/Common/Thread/Event.h new file mode 100644 index 0000000000..ae608a6eb6 --- /dev/null +++ b/Common/Thread/Event.h @@ -0,0 +1,27 @@ +#pragma once + +#include "Common/Thread/ThreadManager.h" +#include +#include + +struct Event : public Waitable { +public: + void Wait() override { + if (triggered_) { + return; + } + std::unique_lock lock; + cond_.wait(lock, [&] { return !triggered_; }); + } + + void Notify() { + std::unique_lock lock; + triggered_ = true; + cond_.notify_one(); + } + +private: + std::condition_variable cond_; + std::mutex mutex_; + bool triggered_ = false; +}; diff --git a/Common/Thread/ParallelLoop.cpp b/Common/Thread/ParallelLoop.cpp index 32ec1aa7fb..bb6868efb8 100644 --- a/Common/Thread/ParallelLoop.cpp +++ b/Common/Thread/ParallelLoop.cpp @@ -7,7 +7,7 @@ public: LoopRangeTask(WaitableCounter *counter, const std::function &loop, int lower, int upper) : counter_(counter), loop_(loop), lower_(lower), upper_(upper) {} - void run() override { + void Run() override { loop_(lower_, upper_); counter_->Count(); } diff --git a/Common/Thread/ParallelLoop.h b/Common/Thread/ParallelLoop.h index 8fa0aac74c..4f82c2a191 100644 --- a/Common/Thread/ParallelLoop.h +++ b/Common/Thread/ParallelLoop.h @@ -8,7 +8,7 @@ #include "Common/Thread/ThreadManager.h" // Kind of like a semaphore I guess. -struct WaitableCounter { +struct WaitableCounter : public Waitable { public: WaitableCounter(int maxValue) : maxValue_(maxValue) {} @@ -19,18 +19,13 @@ public: } } - void Wait() { + void Wait() override { std::unique_lock lock(mutex_); while (count_.load() != maxValue_) { cond_.wait(lock); } } - void WaitAndRelease() { - Wait(); - delete this; - } - int maxValue_; std::atomic count_; std::mutex mutex_; diff --git a/Common/Thread/PrioritizedWorkQueue.cpp b/Common/Thread/PrioritizedWorkQueue.cpp deleted file mode 100644 index 973ef43ce4..0000000000 --- a/Common/Thread/PrioritizedWorkQueue.cpp +++ /dev/null @@ -1,140 +0,0 @@ -#include -#include - -#include "Common/TimeUtil.h" -#include "Common/Thread/PrioritizedWorkQueue.h" - -#include "Common/Log.h" - -PrioritizedWorkQueue::~PrioritizedWorkQueue() { - if (!done_) { - ERROR_LOG(SYSTEM, "PrioritizedWorkQueue destroyed but not done!"); - } -} - -void PrioritizedWorkQueue::Add(PrioritizedWorkQueueItem *item) { - std::lock_guard guard(mutex_); - queue_.push_back(item); - notEmpty_.notify_one(); -} - -void PrioritizedWorkQueue::Stop() { - std::lock_guard guard(mutex_); - done_ = true; - notEmpty_.notify_one(); -} - -void PrioritizedWorkQueue::Flush() { - std::lock_guard guard(mutex_); - int flush_count = 0; - for (auto iter = queue_.begin(); iter != queue_.end(); ++iter) { - delete *iter; - flush_count++; - } - queue_.clear(); - if (flush_count > 0) { - INFO_LOG(SYSTEM, "PrioritizedWorkQueue: Flushed %d un-executed tasks", flush_count); - } -} - -bool PrioritizedWorkQueue::WaitUntilDone(bool all) { - // We'll lock drain this entire time, so make sure you follow that lock ordering. - std::unique_lock guard(drainMutex_); - if (AllItemsDone()) { - return true; - } - - while (!AllItemsDone()) { - drain_.wait(guard); - if (!all) { - // Return whether empty or not, something just drained. - return AllItemsDone(); - } - } - - return true; -} - -void PrioritizedWorkQueue::NotifyDrain() { - std::lock_guard guard(drainMutex_); - drain_.notify_one(); -} - -bool PrioritizedWorkQueue::AllItemsDone() { - std::lock_guard guard(mutex_); - return queue_.empty() && !working_; -} - -// The worker should simply call this in a loop. Will block when appropriate. -PrioritizedWorkQueueItem *PrioritizedWorkQueue::Pop() { - { - std::lock_guard guard(mutex_); - working_ = false; // The thread only calls Pop if it's done. - } - - // Important: make sure mutex_ is not locked while draining. - NotifyDrain(); - - std::unique_lock guard(mutex_); - if (done_) { - return 0; - } - - while (queue_.empty()) { - notEmpty_.wait(guard); - if (done_) { - return 0; - } - } - - // Find the top priority item (lowest value). - float best_prio = std::numeric_limits::infinity(); - std::vector::iterator best = queue_.end(); - for (auto iter = queue_.begin(); iter != queue_.end(); ++iter) { - if ((*iter)->priority() < best_prio) { - best = iter; - best_prio = (*iter)->priority(); - } - } - - if (best != queue_.end()) { - PrioritizedWorkQueueItem *poppedItem = *best; - queue_.erase(best); - working_ = true; // This will be worked on. - return poppedItem; - } else { - // Not really sure how this can happen, but let's be safe. - return 0; - } -} - -// TODO: This feels ugly. Revisit later. - -static std::thread *workThread; - -static void threadfunc(PrioritizedWorkQueue *wq) { - SetCurrentThreadName("PrioQueue"); - while (true) { - PrioritizedWorkQueueItem *item = wq->Pop(); - if (!item) { - if (wq->Done()) - break; - } else { - item->run(); - delete item; - } - } -} - -void ProcessWorkQueueOnThreadWhile(PrioritizedWorkQueue *wq) { - workThread = new std::thread([=](){threadfunc(wq);}); -} - -void StopProcessingWorkQueue(PrioritizedWorkQueue *wq) { - wq->Stop(); - if (workThread) { - workThread->join(); - delete workThread; - } - workThread = nullptr; -} diff --git a/Common/Thread/PrioritizedWorkQueue.h b/Common/Thread/PrioritizedWorkQueue.h deleted file mode 100644 index 59231f48b1..0000000000 --- a/Common/Thread/PrioritizedWorkQueue.h +++ /dev/null @@ -1,65 +0,0 @@ -#pragma once - -#include -#include -#include -#include - -#include "Common/Thread/ThreadUtil.h" - -#include "Common/Common.h" - -// Priorities can change dynamically. -// Try to make priority() fast, it will be called a lot. - -class PrioritizedWorkQueueItem { -public: - PrioritizedWorkQueueItem() {} - virtual ~PrioritizedWorkQueueItem() {} - virtual void run() = 0; - virtual float priority() = 0; // Low priority value = high priority! - -private: - DISALLOW_COPY_AND_ASSIGN(PrioritizedWorkQueueItem); -}; - -class PrioritizedWorkQueue { -public: - PrioritizedWorkQueue() : done_(false), working_(false) {} - ~PrioritizedWorkQueue(); - // Takes ownership. - void Add(PrioritizedWorkQueueItem *item); - - // The worker should simply call this in a loop. Will block when appropriate. - PrioritizedWorkQueueItem *Pop(); - - void Flush(); - bool Done() { return done_; } - void Stop(); - bool WaitUntilDone(bool all = true); - - bool IsWorking() { - return working_; - } - -private: - void NotifyDrain(); - bool AllItemsDone(); - - bool done_; - bool working_; - std::mutex mutex_; - std::mutex drainMutex_; - std::condition_variable notEmpty_; - std::condition_variable drain_; - - std::vector queue_; - - DISALLOW_COPY_AND_ASSIGN(PrioritizedWorkQueue); -}; - - -// Starts up a thread that keeps trying to run this workqueue. -// TODO: This feels ugly. Revisit later. -void ProcessWorkQueueOnThreadWhile(PrioritizedWorkQueue *wq); -void StopProcessingWorkQueue(PrioritizedWorkQueue *wq); diff --git a/Common/Thread/Promise.h b/Common/Thread/Promise.h index a0aac8fc50..c8f2c2ea23 100644 --- a/Common/Thread/Promise.h +++ b/Common/Thread/Promise.h @@ -13,7 +13,7 @@ public: tx_->Release(); } - void run() override { + void Run() override { T *value = fun_(); tx_->Send(value); tx_->Release(); diff --git a/Common/Thread/ThreadManager.cpp b/Common/Thread/ThreadManager.cpp index cc04cb1438..64edf6ed00 100644 --- a/Common/Thread/ThreadManager.cpp +++ b/Common/Thread/ThreadManager.cpp @@ -72,7 +72,7 @@ static void WorkerThreadFunc(GlobalThreadContext *global, ThreadContext *thread) // The task itself takes care of notifying anyone waiting on it. Not the // responsibility of the ThreadManager (although it could be!). if (task) { - task->run(); + task->Run(); delete task; } } @@ -124,6 +124,6 @@ int ThreadManager::GetNumLooperThreads() const { return (int)(global_->threads_.size() / 2); } -void ThreadManager::TryCancelTask(Task *task) { +void ThreadManager::TryCancelTask(uint64_t taskID) { // Do nothing } diff --git a/Common/Thread/ThreadManager.h b/Common/Thread/ThreadManager.h index b3434e8761..b87214363f 100644 --- a/Common/Thread/ThreadManager.h +++ b/Common/Thread/ThreadManager.h @@ -1,5 +1,7 @@ #pragma once +#include + // The new threadpool. // To help future smart scheduling. @@ -8,12 +10,27 @@ enum class TaskType { IO_BLOCKING, }; +// Implement this to make something that you can run on the thread manager. class Task { public: virtual ~Task() {} - virtual void run() = 0; - virtual bool cancellable() { return false; } - virtual void cancel() {} + virtual void Run() = 0; + virtual bool Cancellable() { return false; } + virtual void Cancel() {} + virtual float Priority() { return 1.0f; } + virtual uint64_t id() { return 0; } +}; + +class Waitable { +public: + virtual ~Waitable() {} + + virtual void Wait() = 0; + + void WaitAndRelease() { + Wait(); + delete this; + } }; struct ThreadContext; @@ -29,8 +46,9 @@ public: void EnqueueTaskOnThread(int threadNum, Task *task, TaskType taskType = TaskType::CPU_COMPUTE); // Currently does nothing. It will always be best-effort - maybe it cancels, - // maybe it doesn't. - void TryCancelTask(Task *task); + // maybe it doesn't. Note that the id is the id() returned by the task. You need to make that + // something meaningful yourself. + void TryCancelTask(uint64_t id); // Parallel loops get to use half the threads, // so we still have some worker threads for other tasks. diff --git a/UI/GameInfoCache.cpp b/UI/GameInfoCache.cpp index 64e95df69e..9c9b3dd0df 100644 --- a/UI/GameInfoCache.cpp +++ b/UI/GameInfoCache.cpp @@ -23,7 +23,7 @@ #include #include "Common/GPU/thin3d.h" -#include "Common/Thread/PrioritizedWorkQueue.h" +#include "Common/Thread/ThreadManager.h" #include "Common/File/VFS/VFS.h" #include "Common/File/FileUtil.h" #include "Common/File/Path.h" @@ -214,8 +214,8 @@ bool GameInfo::LoadFromPath(const Path &gamePath) { std::shared_ptr GameInfo::GetFileLoader() { if (filePath_.empty()) { - // Happens when workqueue tries to figure out priorities in PrioritizedWorkQueue::Pop(), - // because priority() calls GetFileLoader()... gnarly. + // Happens when workqueue tries to figure out priorities, + // because Priority() calls GetFileLoader()... gnarly. return fileLoader; } if (!fileLoader) { @@ -331,7 +331,7 @@ static bool ReadVFSToString(const char *filename, std::string *contents, std::mu } -class GameInfoWorkItem : public PrioritizedWorkQueueItem { +class GameInfoWorkItem : public Task { public: GameInfoWorkItem(const Path &gamePath, std::shared_ptr &info) : gamePath_(gamePath), info_(info) { @@ -341,15 +341,18 @@ public: info_->DisposeFileLoader(); } - void run() override { + void Run() override { + // NOTE! DO NOT early-return from this, always use "goto done". It's essential + // that we trigger the event. + if (!info_->LoadFromPath(gamePath_)) { info_->pending = false; - return; + goto done; } // In case of a remote file, check if it actually exists before locking. if (!info_->GetFileLoader()->Exists()) { info_->pending = false; - return; + goto done; } info_->working = true; @@ -371,10 +374,8 @@ public: if (pbp.IsELF()) { goto handleELF; } - ERROR_LOG(LOADER, "invalid pbp '%s'\n", pbpLoader->GetPath().ToVisualString().c_str()); - info_->pending = false; - info_->working = false; - return; + ERROR_LOG(LOADER, "invalid pbp '%s'\n", pbpLoader->GetPath().c_str()); + goto done; } // First, PARAM.SFO. @@ -541,15 +542,11 @@ handleELF: // few files. auto fl = info_->GetFileLoader(); if (!fl) { - info_->pending = false; - info_->working = false; - return; // Happens with UWP currently, TODO... + goto done; } BlockDevice *bd = constructBlockDevice(info_->GetFileLoader().get()); if (!bd) { - info_->pending = false; - info_->working = false; - return; // nothing to do here.. + goto done; } ISOFileSystem umd(&handles, bd); @@ -629,12 +626,14 @@ handleELF: info_->installDataSize = info_->GetInstallDataSizeInBytes(); } +done: info_->pending = false; info_->working = false; + info_->readyEvent.Notify(); // INFO_LOG(SYSTEM, "Completed writing info for %s", info_->GetTitle().c_str()); } - float priority() override { + float Priority() override { auto fl = info_->GetFileLoader(); if (fl && fl->IsRemote()) { // Increase the value so remote info loads after non-remote. @@ -649,7 +648,7 @@ private: DISALLOW_COPY_AND_ASSIGN(GameInfoWorkItem); }; -GameInfoCache::GameInfoCache() : gameInfoWQ_(nullptr) { +GameInfoCache::GameInfoCache() { Init(); } @@ -658,28 +657,15 @@ GameInfoCache::~GameInfoCache() { Shutdown(); } -void GameInfoCache::Init() { - gameInfoWQ_ = new PrioritizedWorkQueue(); - ProcessWorkQueueOnThreadWhile(gameInfoWQ_); -} +void GameInfoCache::Init() {} void GameInfoCache::Shutdown() { CancelAll(); - - if (gameInfoWQ_) { - StopProcessingWorkQueue(gameInfoWQ_); - delete gameInfoWQ_; - gameInfoWQ_ = nullptr; - } } void GameInfoCache::Clear() { CancelAll(); - if (gameInfoWQ_) { - gameInfoWQ_->Flush(); - gameInfoWQ_->WaitUntilDone(); - } info_.clear(); } @@ -706,31 +692,21 @@ void GameInfoCache::FlushBGs() { } void GameInfoCache::PurgeType(IdentifiedFileType fileType) { - if (gameInfoWQ_) - gameInfoWQ_->Flush(); - restart: - for (auto iter = info_.begin(); iter != info_.end(); iter++) { + for (auto iter = info_.begin(); iter != info_.end();) { auto &info = iter->second; - if (!info->working && info->fileType == fileType) { - info_.erase(iter); - goto restart; + info->readyEvent.Wait(); + if (info->fileType == fileType) { + iter = info_.erase(iter); + } else { + iter++; } } } void GameInfoCache::WaitUntilDone(std::shared_ptr &info) { - while (info->pending) { - if (gameInfoWQ_->WaitUntilDone(false)) { - // A true return means everything finished, so bail out. - // This way even if something gets stuck, we won't hang. - break; - } - - // Otherwise, wait again if it's not done... - } + info->readyEvent.Wait(); } - // Runs on the main thread. Only call from render() and similar, not update()! // Can also be called from the audio thread for menu background music. std::shared_ptr GameInfoCache::GetInfo(Draw::DrawContext *draw, const Path &gamePath, int wantFlags) { @@ -776,11 +752,11 @@ std::shared_ptr GameInfoCache::GetInfo(Draw::DrawContext *draw, const } GameInfoWorkItem *item = new GameInfoWorkItem(gamePath, info); - gameInfoWQ_->Add(item); + g_threadManager.EnqueueTask(item); // Don't re-insert if we already have it. if (info_.find(pathStr) == info_.end()) - info_[pathStr] = std::shared_ptr(info); + info_[pathStr] = info; return info; } diff --git a/UI/GameInfoCache.h b/UI/GameInfoCache.h index f2abd58c7d..c39408b231 100644 --- a/UI/GameInfoCache.h +++ b/UI/GameInfoCache.h @@ -23,6 +23,7 @@ #include #include +#include "Common/Thread/Event.h" #include "Core/ELF/ParamSFO.h" #include "Common/File/Path.h" #include "UI/TextureUtil.h" @@ -31,7 +32,6 @@ namespace Draw { class DrawContext; class Texture; } -class PrioritizedWorkQueue; // A GameInfo holds information about a game, and also lets you do things that the VSH // does on the PSP, namely checking for and deleting savedata, and similar things. @@ -140,9 +140,12 @@ public: u64 gameSize = 0; u64 saveDataSize = 0; u64 installDataSize = 0; + std::atomic pending{}; std::atomic working{}; + Event readyEvent; + protected: // Note: this can change while loading, use GetTitle(). std::string title; @@ -182,9 +185,6 @@ private: // Maps ISO path to info. Need to use shared_ptr as we can return these pointers - // and if they get destructed while being in use, that's bad. std::map > info_; - - // Work queue and management - PrioritizedWorkQueue *gameInfoWQ_; }; // This one can be global, no good reason not to. diff --git a/UWP/CommonUWP/CommonUWP.vcxproj b/UWP/CommonUWP/CommonUWP.vcxproj index 4323046089..feb79abb97 100644 --- a/UWP/CommonUWP/CommonUWP.vcxproj +++ b/UWP/CommonUWP/CommonUWP.vcxproj @@ -486,7 +486,7 @@ - + @@ -599,7 +599,7 @@ - + diff --git a/UWP/CommonUWP/CommonUWP.vcxproj.filters b/UWP/CommonUWP/CommonUWP.vcxproj.filters index 37e4ef9af2..5755c6567a 100644 --- a/UWP/CommonUWP/CommonUWP.vcxproj.filters +++ b/UWP/CommonUWP/CommonUWP.vcxproj.filters @@ -180,7 +180,7 @@ Input - + Thread @@ -460,7 +460,7 @@ Input - + Thread diff --git a/android/jni/Android.mk b/android/jni/Android.mk index 6d96159bcd..e2f73f3b1d 100644 --- a/android/jni/Android.mk +++ b/android/jni/Android.mk @@ -285,7 +285,11 @@ EXEC_AND_LIB_FILES := \ $(SRC)/Common/Net/WebsocketServer.cpp \ $(SRC)/Common/Profiler/Profiler.cpp \ $(SRC)/Common/System/Display.cpp \ +<<<<<<< HEAD $(SRC)/Common/Thread/PrioritizedWorkQueue.cpp \ +======= + $(SRC)/Common/Thread/Executor.cpp \ +>>>>>>> 82282c7a3 (Get rid of the PrioritizedWorkQueue. Instead just queue tasks on the ThreadManager.) $(SRC)/Common/Thread/ThreadUtil.cpp \ $(SRC)/Common/Thread/ThreadManager.cpp \ $(SRC)/Common/Thread/ParallelLoop.cpp \ diff --git a/libretro/Makefile.common b/libretro/Makefile.common index 27cf1d35cb..d4e45f7803 100644 --- a/libretro/Makefile.common +++ b/libretro/Makefile.common @@ -278,7 +278,6 @@ SOURCES_CXX += \ $(COMMONDIR)/Render/TextureAtlas.cpp \ $(COMMONDIR)/Serialize/Serializer.cpp \ $(COMMONDIR)/Thread/ThreadUtil.cpp \ - $(COMMONDIR)/Thread/PrioritizedWorkQueue.cpp \ $(COMMONDIR)/Thread/ParallelLoop.cpp \ $(COMMONDIR)/Thread/ThreadManager.cpp \ $(COMMONDIR)/UI/Root.cpp \