From f9bab64bdf1bd674df9ed39b22c3730b1c6e30dc Mon Sep 17 00:00:00 2001 From: "Unknown W. Brackets" Date: Sun, 5 Dec 2021 19:06:18 -0800 Subject: [PATCH 1/3] Android: Optimize content URI exists check. --- Core/FileLoaders/LocalFileLoader.cpp | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/Core/FileLoaders/LocalFileLoader.cpp b/Core/FileLoaders/LocalFileLoader.cpp index 2e2caeff74..19e5be2e59 100644 --- a/Core/FileLoaders/LocalFileLoader.cpp +++ b/Core/FileLoaders/LocalFileLoader.cpp @@ -60,9 +60,9 @@ LocalFileLoader::LocalFileLoader(const Path &filename) #if PPSSPP_PLATFORM(ANDROID) if (filename.Type() == PathType::CONTENT_URI) { int fd = Android_OpenContentUriFd(filename.ToString(), Android_OpenContentUriMode::READ); - VERBOSE_LOG(SYSTEM, "Fd %d for content URI: '%s'", fd, filename.c_str()); + VERBOSE_LOG(SYSTEM, "LocalFileLoader Fd %d for content URI: '%s'", fd, filename.c_str()); if (fd < 0) { - ERROR_LOG(FILESYS, "LoadFileLoader failed to open content URI: '%s'", filename.c_str()); + ERROR_LOG(FILESYS, "LocalFileLoader failed to open content URI: '%s'", filename.c_str()); return; } fd_ = fd; @@ -118,23 +118,26 @@ LocalFileLoader::~LocalFileLoader() { } bool LocalFileLoader::Exists() { - // If we couldn't open it for reading, we say it does not exist. + // If we opened it for reading, it must exist. Done. #ifndef _WIN32 if (isOpenedByFd_) { + // As an optimization, if we already tried and failed, quickly return. + // This is used because Android Content URIs are so slow. return fd_ != -1; } - if (fd_ != -1 || IsDirectory()) { + if (fd_ != -1) + return true; #else - if (handle_ != INVALID_HANDLE_VALUE || IsDirectory()) { + if (handle_ != INVALID_HANDLE_VALUE) + return true; #endif - File::FileInfo info; - if (File::GetFileInfo(filename_, &info)) { - return info.exists; - } else { - return false; - } + + File::FileInfo info; + if (File::GetFileInfo(filename_, &info)) { + return info.exists; + } else { + return false; } - return false; } bool LocalFileLoader::IsDirectory() { From 8b5173350fcdcdc690ef48c48e4dc5980871328c Mon Sep 17 00:00:00 2001 From: "Unknown W. Brackets" Date: Sun, 5 Dec 2021 19:22:13 -0800 Subject: [PATCH 2/3] ThreadManager: Make TaskType a property of a Task. --- Common/Thread/ParallelLoop.cpp | 8 ++++++-- Common/Thread/Promise.h | 11 ++++++++--- Common/Thread/ThreadManager.cpp | 6 +++--- Common/Thread/ThreadManager.h | 5 +++-- Core/TextureReplacer.cpp | 6 +++++- UI/GameInfoCache.cpp | 6 +++++- 6 files changed, 30 insertions(+), 12 deletions(-) diff --git a/Common/Thread/ParallelLoop.cpp b/Common/Thread/ParallelLoop.cpp index 0212a84b8a..2f00664618 100644 --- a/Common/Thread/ParallelLoop.cpp +++ b/Common/Thread/ParallelLoop.cpp @@ -9,6 +9,10 @@ public: LoopRangeTask(WaitableCounter *counter, const std::function &loop, int lower, int upper) : counter_(counter), loop_(loop), lower_(lower), upper_(upper) {} + TaskType Type() const override { + return TaskType::CPU_COMPUTE; + } + void Run() override { loop_(lower_, upper_); counter_->Count(); @@ -34,7 +38,7 @@ WaitableCounter *ParallelRangeLoopWaitable(ThreadManager *threadMan, const std:: } else if (range <= minSize) { // Single background task. WaitableCounter *waitableCounter = new WaitableCounter(1); - threadMan->EnqueueTaskOnThread(0, new LoopRangeTask(waitableCounter, loop, lower, upper), TaskType::CPU_COMPUTE); + threadMan->EnqueueTaskOnThread(0, new LoopRangeTask(waitableCounter, loop, lower, upper)); return waitableCounter; } else { // Split the range between threads. Allow for some fractional bits. @@ -61,7 +65,7 @@ WaitableCounter *ParallelRangeLoopWaitable(ThreadManager *threadMan, const std:: // Let's do the stragglers on the current thread. break; } - threadMan->EnqueueTaskOnThread(i, new LoopRangeTask(waitableCounter, loop, start, end), TaskType::CPU_COMPUTE); + threadMan->EnqueueTaskOnThread(i, new LoopRangeTask(waitableCounter, loop, start, end)); counter += delta; if ((counter >> fractionalBits) >= upper) { break; diff --git a/Common/Thread/Promise.h b/Common/Thread/Promise.h index f69bb791e9..cd057631ef 100644 --- a/Common/Thread/Promise.h +++ b/Common/Thread/Promise.h @@ -9,13 +9,17 @@ template class PromiseTask : public Task { public: - PromiseTask(std::function fun, Mailbox *tx) : fun_(fun), tx_(tx) { + PromiseTask(std::function fun, Mailbox *tx, TaskType t) : fun_(fun), tx_(tx), type_(t) { tx_->AddRef(); } ~PromiseTask() { tx_->Release(); } + TaskType Type() const override { + return type_; + } + void Run() override { T value = fun_(); tx_->Send(value); @@ -23,6 +27,7 @@ public: std::function fun_; Mailbox *tx_; + TaskType type_; }; // Represents pending or actual data. @@ -38,8 +43,8 @@ public: Promise *promise = new Promise(); promise->rx_ = mailbox; - PromiseTask *task = new PromiseTask(fun, mailbox); - threadman->EnqueueTask(task, taskType); + PromiseTask *task = new PromiseTask(fun, mailbox, taskType); + threadman->EnqueueTask(task); return promise; } diff --git a/Common/Thread/ThreadManager.cpp b/Common/Thread/ThreadManager.cpp index 5cdd23d64e..1a48645357 100644 --- a/Common/Thread/ThreadManager.cpp +++ b/Common/Thread/ThreadManager.cpp @@ -172,12 +172,12 @@ void ThreadManager::Init(int numRealCores, int numLogicalCoresPerCpu) { } } -void ThreadManager::EnqueueTask(Task *task, TaskType taskType) { +void ThreadManager::EnqueueTask(Task *task) { _assert_msg_(IsInitialized(), "ThreadManager not initialized"); int maxThread; int threadOffset = 0; - if (taskType == TaskType::CPU_COMPUTE) { + if (task->Type() == TaskType::CPU_COMPUTE) { // only the threads reserved for heavy compute. maxThread = numComputeThreads_; threadOffset = 0; @@ -219,7 +219,7 @@ void ThreadManager::EnqueueTask(Task *task, TaskType taskType) { chosenThread->cond.notify_one(); } -void ThreadManager::EnqueueTaskOnThread(int threadNum, Task *task, TaskType taskType) { +void ThreadManager::EnqueueTaskOnThread(int threadNum, Task *task) { _assert_msg_(threadNum >= 0 && threadNum < (int)global_->threads_.size(), "Bad threadnum or not initialized"); ThreadContext *thread = global_->threads_[threadNum]; diff --git a/Common/Thread/ThreadManager.h b/Common/Thread/ThreadManager.h index a56184c44f..f6ddd842b2 100644 --- a/Common/Thread/ThreadManager.h +++ b/Common/Thread/ThreadManager.h @@ -14,6 +14,7 @@ enum class TaskType { class Task { public: virtual ~Task() {} + virtual TaskType Type() const = 0; virtual void Run() = 0; virtual bool Cancellable() { return false; } virtual void Cancel() {} @@ -44,8 +45,8 @@ public: // It gets even trickier when you think about mobile chips with BIG/LITTLE, but we'll // just ignore it and let the OS handle it. void Init(int numCores, int numLogicalCoresPerCpu); - void EnqueueTask(Task *task, TaskType taskType); - void EnqueueTaskOnThread(int threadNum, Task *task, TaskType taskType); + void EnqueueTask(Task *task); + void EnqueueTaskOnThread(int threadNum, Task *task); void Teardown(); bool IsInitialized() const; diff --git a/Core/TextureReplacer.cpp b/Core/TextureReplacer.cpp index 3d40e302d0..f4a2e72828 100644 --- a/Core/TextureReplacer.cpp +++ b/Core/TextureReplacer.cpp @@ -785,6 +785,10 @@ public: ReplacedTextureTask(ReplacedTexture &tex, LimitedWaitable *w) : tex_(tex), waitable_(w) { } + TaskType Type() const override { + return TaskType::IO_BLOCKING; + } + void Run() override { tex_.Prepare(); waitable_->Notify(); @@ -815,7 +819,7 @@ bool ReplacedTexture::IsReady(double budget) { if (g_Config.bReplaceTexturesAllowLate) { threadWaitable_ = new LimitedWaitable(); - g_threadManager.EnqueueTask(new ReplacedTextureTask(*this, threadWaitable_), TaskType::IO_BLOCKING); + g_threadManager.EnqueueTask(new ReplacedTextureTask(*this, threadWaitable_)); if (threadWaitable_->WaitFor(budget)) { threadWaitable_->WaitAndRelease(); diff --git a/UI/GameInfoCache.cpp b/UI/GameInfoCache.cpp index e3bbfb34d0..83743e4090 100644 --- a/UI/GameInfoCache.cpp +++ b/UI/GameInfoCache.cpp @@ -340,6 +340,10 @@ public: info_->readyEvent.Notify(); } + TaskType Type() const override { + return TaskType::IO_BLOCKING; + } + void Run() override { // An early-return will result in the destructor running, where we can set // flags like working and pending. @@ -738,7 +742,7 @@ std::shared_ptr GameInfoCache::GetInfo(Draw::DrawContext *draw, const } GameInfoWorkItem *item = new GameInfoWorkItem(gamePath, info); - g_threadManager.EnqueueTask(item, TaskType::IO_BLOCKING); + g_threadManager.EnqueueTask(item); // Don't re-insert if we already have it. if (info_.find(pathStr) == info_.end()) From f9a7ad3e3d19af3ad5461d4ec8011f89445e416e Mon Sep 17 00:00:00 2001 From: "Unknown W. Brackets" Date: Sun, 5 Dec 2021 20:30:37 -0800 Subject: [PATCH 3/3] ThreadManager: Use separate pool for IO blocking. This prevents starving the compute pool (which may be used very regularly parallel loops or other tasks) if the IO operations are slow. --- Common/Thread/ThreadManager.cpp | 117 +++++++++++++++++++++----------- 1 file changed, 78 insertions(+), 39 deletions(-) diff --git a/Common/Thread/ThreadManager.cpp b/Common/Thread/ThreadManager.cpp index 1a48645357..588e88d520 100644 --- a/Common/Thread/ThreadManager.cpp +++ b/Common/Thread/ThreadManager.cpp @@ -21,12 +21,14 @@ // is not fair. However, we ignore that for now. const int MAX_CORES_TO_USE = 16; -const int EXTRA_THREADS = 4; // For I/O limited tasks +const int MIN_IO_BLOCKING_THREADS = 4; struct GlobalThreadContext { std::mutex mutex; // associated with each respective condition variable - std::deque queue; - std::atomic queue_size; + std::deque compute_queue; + std::atomic compute_queue_size; + std::deque io_queue; + std::atomic io_queue_size; std::vector threads_; std::atomic roundRobin; @@ -38,13 +40,15 @@ struct ThreadContext { std::mutex mutex; // protects the local queue. std::atomic queue_size; int index; + TaskType type; std::atomic cancelled; std::atomic private_single; std::deque private_queue; }; ThreadManager::ThreadManager() : global_(new GlobalThreadContext()) { - global_->queue_size = 0; + global_->compute_queue_size = 0; + global_->io_queue_size = 0; global_->roundRobin = 0; } @@ -60,19 +64,23 @@ void ThreadManager::Teardown() { } // Purge any cancellable tasks while the threads shut down. - bool done = false; - while (!done) { - done = true; + if (global_->compute_queue_size > 0 || global_->io_queue_size > 0) { + auto drainQueue = [&](std::deque &queue, std::atomic &size) { + for (auto it = queue.begin(); it != queue.end(); ++it) { + if (TeardownTask(*it, false)) { + queue.erase(it); + size--; + return false; + } + } + return true; + }; std::unique_lock lock(global_->mutex); - for (auto it = global_->queue.begin(); it != global_->queue.end(); ++it) { - if (TeardownTask(*it, false)) { - global_->queue.erase(it); - global_->queue_size--; - done = false; - break; - } - } + while (!drainQueue(global_->compute_queue, global_->compute_queue_size)) + continue; + while (!drainQueue(global_->io_queue, global_->io_queue_size)) + continue; } for (ThreadContext *&threadCtx : global_->threads_) { @@ -86,7 +94,7 @@ void ThreadManager::Teardown() { } global_->threads_.clear(); - if (global_->queue_size > 0) { + if (global_->compute_queue_size > 0 || global_->io_queue_size > 0) { WARN_LOG(SYSTEM, "ThreadManager::Teardown() with tasks still enqueued"); } } @@ -102,8 +110,15 @@ bool ThreadManager::TeardownTask(Task *task, bool enqueue) { } if (enqueue) { - global_->queue.push_back(task); - global_->queue_size++; + if (task->Type() == TaskType::CPU_COMPUTE) { + global_->compute_queue.push_back(task); + global_->compute_queue_size++; + } else if (task->Type() == TaskType::CPU_COMPUTE) { + global_->io_queue.push_back(task); + global_->io_queue_size++; + } else { + _assert_(false); + } } return false; } @@ -112,17 +127,26 @@ static void WorkerThreadFunc(GlobalThreadContext *global, ThreadContext *thread) char threadName[16]; snprintf(threadName, sizeof(threadName), "PoolWorker %d", thread->index); SetCurrentThreadName(threadName); + + const bool isCompute = thread->type == TaskType::CPU_COMPUTE; + const auto global_queue_size = [isCompute, &global]() -> int { + return isCompute ? global->compute_queue_size.load() : global->io_queue_size.load(); + }; + while (!thread->cancelled) { Task *task = thread->private_single.exchange(nullptr); // Check the global queue first, then check the private queue and wait if there's nothing to do. - if (!task && global->queue_size.load() > 0) { + if (!task && global_queue_size() > 0) { // Grab one from the global queue if there is any. std::unique_lock lock(global->mutex); - if (!global->queue.empty()) { - task = global->queue.front(); - global->queue.pop_front(); - global->queue_size--; + auto &queue = isCompute ? global->compute_queue : global->io_queue; + auto &queue_size = isCompute ? global->compute_queue_size : global->io_queue_size; + + if (!queue.empty()) { + task = queue.front(); + queue.pop_front(); + queue_size--; // We are processing one now, so mark that. thread->queue_size++; @@ -132,12 +156,19 @@ static void WorkerThreadFunc(GlobalThreadContext *global, ThreadContext *thread) if (!task) { std::unique_lock lock(thread->mutex); // We must check both queue and single again, while locked. + bool wait = true; if (!thread->private_queue.empty()) { task = thread->private_queue.front(); thread->private_queue.pop_front(); - } else if (!thread->private_single && !thread->cancelled && global->queue_size.load() == 0) { - thread->cond.wait(lock); + wait = false; + } else if (thread->private_single || thread->cancelled) { + wait = false; + } else { + wait = global_queue_size() == 0; } + + if (wait) + thread->cond.wait(lock); } // The task itself takes care of notifying anyone waiting on it. Not the // responsibility of the ThreadManager (although it could be!). @@ -157,7 +188,8 @@ void ThreadManager::Init(int numRealCores, int numLogicalCoresPerCpu) { } numComputeThreads_ = std::min(numRealCores * numLogicalCoresPerCpu, MAX_CORES_TO_USE); - int numThreads = numComputeThreads_ + EXTRA_THREADS; + // Double it for the IO blocking threads. + int numThreads = numComputeThreads_ + std::max(MIN_IO_BLOCKING_THREADS, numComputeThreads_); numThreads_ = numThreads; INFO_LOG(SYSTEM, "ThreadManager::Init(compute threads: %d, all: %d)", numComputeThreads_, numThreads_); @@ -166,6 +198,7 @@ void ThreadManager::Init(int numRealCores, int numLogicalCoresPerCpu) { ThreadContext *thread = new ThreadContext(); thread->cancelled.store(false); thread->private_single.store(nullptr); + thread->type = i < numComputeThreads_ ? TaskType::CPU_COMPUTE : TaskType::IO_BLOCKING; thread->thread = std::thread(&WorkerThreadFunc, global_, thread); thread->index = i; global_->threads_.push_back(thread); @@ -175,24 +208,21 @@ void ThreadManager::Init(int numRealCores, int numLogicalCoresPerCpu) { void ThreadManager::EnqueueTask(Task *task) { _assert_msg_(IsInitialized(), "ThreadManager not initialized"); + int minThread; int maxThread; - int threadOffset = 0; if (task->Type() == TaskType::CPU_COMPUTE) { // only the threads reserved for heavy compute. + minThread = 0; maxThread = numComputeThreads_; - threadOffset = 0; } else { - // any free thread + // Only IO blocking threads (to avoid starving compute threads.) + minThread = numComputeThreads_; maxThread = numThreads_; - threadOffset = numComputeThreads_; } // Find a thread with no outstanding work. - int threadNum = threadOffset; - for (int i = 0; i < maxThread; i++, threadNum++) { - if (threadNum >= global_->threads_.size()) { - threadNum = 0; - } + _assert_(maxThread <= global_->threads_.size()); + for (int threadNum = minThread; threadNum < maxThread; threadNum++) { ThreadContext *thread = global_->threads_[threadNum]; if (thread->queue_size.load() == 0) { std::unique_lock lock(thread->mutex); @@ -208,13 +238,22 @@ void ThreadManager::EnqueueTask(Task *task) { // Not particularly scientific, but hopefully we should not run into this too much. { std::unique_lock lock(global_->mutex); - global_->queue.push_back(task); - global_->queue_size++; + if (task->Type() == TaskType::CPU_COMPUTE) { + global_->compute_queue.push_back(task); + global_->compute_queue_size++; + } else if (task->Type() == TaskType::IO_BLOCKING) { + global_->io_queue.push_back(task); + global_->io_queue_size++; + } else { + _assert_(false); + } } - // Lock the thread to ensure it gets the message. int chosenIndex = global_->roundRobin++; - ThreadContext *&chosenThread = global_->threads_[chosenIndex % maxThread]; + chosenIndex = minThread + (chosenIndex % (maxThread - minThread)); + ThreadContext *&chosenThread = global_->threads_[chosenIndex]; + + // Lock the thread to ensure it gets the message. std::unique_lock lock(chosenThread->mutex); chosenThread->cond.notify_one(); }