Merge pull request #15205 from unknownbrackets/android-content-hang

ThreadManager: Use separate pool for IO blocking

Common/Thread/ParallelLoop.cpp

@@ -9,6 +9,10 @@ public:
 	LoopRangeTask(WaitableCounter *counter, const std::function<void(int, int)> &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;

Common/Thread/Promise.h

@@ -9,13 +9,17 @@
 template<class T>
 class PromiseTask : public Task {
 public:
-	PromiseTask(std::function<T ()> fun, Mailbox<T> *tx) : fun_(fun), tx_(tx) {
+	PromiseTask(std::function<T ()> fun, Mailbox<T> *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<T ()> fun_;
 	Mailbox<T> *tx_;
+	TaskType type_;
 };
 
 // Represents pending or actual data.
@@ -38,8 +43,8 @@ public:
 		Promise<T> *promise = new Promise<T>();
 		promise->rx_ = mailbox;
 
-		PromiseTask<T> *task = new PromiseTask<T>(fun, mailbox);
-		threadman->EnqueueTask(task, taskType);
+		PromiseTask<T> *task = new PromiseTask<T>(fun, mailbox, taskType);
+		threadman->EnqueueTask(task);
 		return promise;
 	}
 

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<Task *> queue;
-	std::atomic<int> queue_size;
+	std::deque<Task *> compute_queue;
+	std::atomic<int> compute_queue_size;
+	std::deque<Task *> io_queue;
+	std::atomic<int> io_queue_size;
 	std::vector<ThreadContext *> threads_;
 
 	std::atomic<int> roundRobin;
@@ -38,13 +40,15 @@ struct ThreadContext {
 	std::mutex mutex; // protects the local queue.
 	std::atomic<int> queue_size;
 	int index;
+	TaskType type;
 	std::atomic<bool> cancelled;
 	std::atomic<Task *> private_single;
 	std::deque<Task *> 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<Task *> &queue, std::atomic<int> &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<std::mutex> 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<std::mutex> 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<std::mutex> 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,33 +198,31 @@ 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);
 	}
 }
 
-void ThreadManager::EnqueueTask(Task *task, TaskType taskType) {
+void ThreadManager::EnqueueTask(Task *task) {
 	_assert_msg_(IsInitialized(), "ThreadManager not initialized");
 
+	int minThread;
 	int maxThread;
-	int threadOffset = 0;
-	if (taskType == TaskType::CPU_COMPUTE) {
+	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<std::mutex> lock(thread->mutex);
@@ -208,18 +238,27 @@ void ThreadManager::EnqueueTask(Task *task, TaskType taskType) {
 	// Not particularly scientific, but hopefully we should not run into this too much.
 	{
 		std::unique_lock<std::mutex> 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<std::mutex> lock(chosenThread->mutex);
 	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];
 

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;

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() {

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();

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<GameInfo> 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())