Differential D4730 Diff 20833 source/ps/TaskManager.cpp

Changeset View

Standalone View

source/ps/TaskManager.cpp

Show First 20 Lines • Show All 55 Lines • ▼ Show 20 Lines
}		}

} // anonymous namespace		} // anonymous namespace

std::unique_ptr<TaskManager> g_TaskManager;		std::unique_ptr<TaskManager> g_TaskManager;

class Thread;		class Thread;

using QueueItem = std::function<void()>;		using QueueItem = PackagedTask;

/**		/**
* Light wrapper around std::thread. Ensures Join has been called.		* Light wrapper around std::thread. Ensures Join has been called.
*/		*/
class Thread		class Thread
{		{
public:		public:
Thread() = default;		Thread() = default;
▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Lines	public:
*/		*/
void SetupWorkers(size_t numberOfWorkers);		void SetupWorkers(size_t numberOfWorkers);

/**		/**
* Push a task on the global queue.		* Push a task on the global queue.
* Takes ownership of @a task.		* Takes ownership of @a task.
* May be called from any thread.		* May be called from any thread.
*/		*/
void PushTask(std::function<void()>&& task, TaskPriority priority);		void PushTask(QueueItem&& task, TaskPriority priority);

protected:		protected:
void ClearQueue();		void ClearQueue();

template<TaskPriority Priority>		template<TaskPriority Priority>
bool PopTask(std::function<void()>& taskOut);		bool PopTask(QueueItem& taskOut);

// Back reference (keep this first).		// Back reference (keep this first).
TaskManager& m_TaskManager;		TaskManager& m_TaskManager;

std::atomic<bool> m_HasWork = false;		std::atomic<bool> m_HasWork = false;
std::atomic<bool> m_HasLowPriorityWork = false;		std::atomic<bool> m_HasLowPriorityWork = false;
std::mutex m_GlobalMutex;		std::mutex m_GlobalMutex;
std::mutex m_GlobalLowPriorityMutex;		std::mutex m_GlobalLowPriorityMutex;
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines	void TaskManager::Impl::ClearQueue()
}		}
}		}

size_t TaskManager::GetNumberOfWorkers() const		size_t TaskManager::GetNumberOfWorkers() const
{		{
return m->m_Workers.size();		return m->m_Workers.size();
}		}

void TaskManager::DoPushTask(std::function<void()>&& task, TaskPriority priority)		void TaskManager::DoPushTask(QueueItem&& task, TaskPriority priority)
{		{
m->PushTask(std::move(task), priority);		m->PushTask(std::move(task), priority);
}		}

void TaskManager::Impl::PushTask(std::function<void()>&& task, TaskPriority priority)		void TaskManager::Impl::PushTask(QueueItem&& task, TaskPriority priority)
{		{
std::mutex& mutex = priority == TaskPriority::NORMAL ? m_GlobalMutex : m_GlobalLowPriorityMutex;		std::mutex& mutex = priority == TaskPriority::NORMAL ? m_GlobalMutex : m_GlobalLowPriorityMutex;
std::deque<QueueItem>& queue = priority == TaskPriority::NORMAL ? m_GlobalQueue : m_GlobalLowPriorityQueue;		std::deque<QueueItem>& queue = priority == TaskPriority::NORMAL ? m_GlobalQueue : m_GlobalLowPriorityQueue;
std::atomic<bool>& hasWork = priority == TaskPriority::NORMAL ? m_HasWork : m_HasLowPriorityWork;		std::atomic<bool>& hasWork = priority == TaskPriority::NORMAL ? m_HasWork : m_HasLowPriorityWork;
{		{
std::lock_guard<std::mutex> lock(mutex);		std::lock_guard<std::mutex> lock(mutex);
queue.emplace_back(std::move(task));		queue.emplace_back(std::move(task));
hasWork = true;		hasWork = true;
}		}

for (WorkerThread& worker : m_Workers)		for (WorkerThread& worker : m_Workers)
worker.Wake();		worker.Wake();
}		}

template<TaskPriority Priority>		template<TaskPriority Priority>
bool TaskManager::Impl::PopTask(std::function<void()>& taskOut)		bool TaskManager::Impl::PopTask(QueueItem& taskOut)
{		{
std::mutex& mutex = Priority == TaskPriority::NORMAL ? m_GlobalMutex : m_GlobalLowPriorityMutex;		std::mutex& mutex = Priority == TaskPriority::NORMAL ? m_GlobalMutex : m_GlobalLowPriorityMutex;
std::deque<QueueItem>& queue = Priority == TaskPriority::NORMAL ? m_GlobalQueue : m_GlobalLowPriorityQueue;		std::deque<QueueItem>& queue = Priority == TaskPriority::NORMAL ? m_GlobalQueue : m_GlobalLowPriorityQueue;
std::atomic<bool>& hasWork = Priority == TaskPriority::NORMAL ? m_HasWork : m_HasLowPriorityWork;		std::atomic<bool>& hasWork = Priority == TaskPriority::NORMAL ? m_HasWork : m_HasLowPriorityWork;
		jprahmanUnsubmitted Not Done Inline Actions Unrelated to this diff, but may be worth separately pulling these three fields into a single struct and having two copies of that struct (one per priority level), and then only having the ternary operator applied once to fetch the correct instance of that struct. jprahman: Unrelated to this diff, but may be worth separately pulling these three fields into a single…

// Particularly critical section since we're locking the global queue.		// Particularly critical section since we're locking the global queue.
std::lock_guard<std::mutex> globalLock(mutex);		std::lock_guard<std::mutex> globalLock(mutex);
if (!queue.empty())		if (!queue.empty())
{		{
taskOut = std::move(queue.front());		taskOut = std::move(queue.front());
queue.pop_front();		queue.pop_front();
hasWork = !queue.empty();		hasWork = !queue.empty();
Show All 39 Lines
{		{
// The profiler does better if the names are unique.		// The profiler does better if the names are unique.
static std::atomic<int> n = 0;		static std::atomic<int> n = 0;
std::string name = "Task Mgr #" + std::to_string(n++);		std::string name = "Task Mgr #" + std::to_string(n++);
debug_SetThreadName(name.c_str());		debug_SetThreadName(name.c_str());
g_Profiler2.RegisterCurrentThread(name);		g_Profiler2.RegisterCurrentThread(name);


std::function<void()> task;		QueueItem task;
bool hasTask = false;		bool hasTask = false;
std::unique_lock<std::mutex> lock(m_Mutex, std::defer_lock);		std::unique_lock<std::mutex> lock(m_Mutex, std::defer_lock);
while (!m_Kill)		while (!m_Kill)
{		{
lock.lock();		lock.lock();
m_ConditionVariable.wait(lock, [this](){		m_ConditionVariable.wait(lock, [this](){
return m_Kill \|\| m_TaskManager.m_HasWork \|\| m_TaskManager.m_HasLowPriorityWork;		return m_Kill \|\| m_TaskManager.m_HasWork \|\| m_TaskManager.m_HasLowPriorityWork;
});		});
Show All 22 Lines