Differential D3848 Diff 17559 source/ps/ThreadPool.cpp

Changeset View

Standalone View

View Options

source/ps/ThreadPool.cpp

This file was added.

* This file is part of 0 A.D.

* 0 A.D. is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, either version 2 of the License, or

* (at your option) any later version.

* 0 A.D. is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

* You should have received a copy of the GNU General Public License

* along with 0 A.D. If not, see <http://www.gnu.org/licenses/>.

#include "precompiled.h"

#include "ThreadPool.h"

#include "lib/debug.h"

#include "maths/MathUtil.h"

#include "ps/CLogger.h"

#include "ps/ConfigDB.h"

#include "ps/Threading.h"

#include "ps/ThreadUtil.h"

#include "ps/Profiler2.h"

#include <condition_variable>

StanUnsubmitted

Not Done

#include "ps/Profiler2.h"

+ #include <atomic>

+ #include <algorithm>

+ #include <condition_variable>

+ #include <deque>

+ #include <functional>

#include <memory>

+ #include <mutex>

#include <list>

+ #include <string>

#include <thread>

CThreadPool g_ThreadPool;

Stan:

#include <deque>

#include <functional>

#include <memory>

#include <mutex>

#include <list>

StanUnsubmitted

Not Done

#include "ps/Profiler2.h"

#include <condition_variable>

+ #include <deque>

#include <functional>

#include <memory>

#include <mutex>

#include <list>

#include <thread>

+ #include <vector>

#define FORCE_SINGLE_THREAD 0

Stan:

#include <thread>

StanUnsubmitted

Not Done

Shouldn't that be detected at runtime when host has one core?. Constexpr?

Stan: Shouldn't that be detected at runtime when host has one core?. Constexpr?

wraitiiAuthorUnsubmitted

Done

This is intended for debugging only, I might drop it in the final diff.

wraitii: This is intended for debugging only, I might drop it in the final diff.

StanUnsubmitted

Not Done

Still there :P

Stan: Still there :P

namespace ThreadPool

{

/**

* Minimum number of thread pool workers.

static constexpr size_t MIN_THREADS = 3;

/**

* Maximum number of thread pool workers.

static constexpr size_t MAX_THREADS = 32;

std::unique_ptr<TaskManager> g_TaskManager;

class Thread;

struct QueueItem

{

vladislavbelovUnsubmitted

Not Done

It's not IThread, it's ThreadOwner or ThreadHolder according to its responsibility.

vladislavbelov: It's not `IThread`, it's `ThreadOwner` or `ThreadHolder` according to its responsibility.

QueueItem(std::function<void()>&& f, WorkerAffinity a) : m_Task(std::move(f)), m_Affinity(a) {}

std::function<void()> m_Task;

WorkerAffinity m_Affinity;

StanUnsubmitted

Not Done

enum class?

Stan: enum class?

wraitiiAuthorUnsubmitted

Done

Annoying for bitfields.

wraitii: Annoying for bitfields.

StanUnsubmitted

Not Done

Could be clamp?

Stan: Could be clamp?

};

}

/**

* Light wrapper around std::thread. Ensures Join has been called.

class ThreadPool::Thread

{

public:

Thread() = default;

Thread(const Thread&) = delete;

Thread(Thread&&) = delete;

wraitiiAuthorUnsubmitted

Done

I think the predicate version actually checks the predicate before waiting, making this a bit redundant - need to check.

wraitii: I think the predicate version actually checks the predicate before waiting, making this a bit…

StanUnsubmitted

Not Done

virtual see msvc output.

Stan: virtual see msvc output.

wraitiiAuthorUnsubmitted

Done

Msvc is wrong here and it annoys me :P

I think I'll just hide the warning.

wraitii: Msvc is wrong here and it annoys me :P I think I'll just hide the warning.

StanUnsubmitted

Done

Care to explain for me? :)

Stan: Care to explain for me? :)

wraitiiAuthorUnsubmitted

Done

Since the destructor is protected, it doesn't need to be virtual - you won't ever call the destructor of a pointer-to-the-base-class-polymorphically-pointing-to-a-derived-class, which is why virtual would be needed (so it's actually the derived class' destructor that's called).

See https://docs.microsoft.com/en-us/cpp/code-quality/c26436?view=msvc-160

I suspect VS2019 wouldn't warn, like gcc and clang.

wraitii: Since the destructor is protected, it doesn't need to be virtual - you won't ever call the…

template<typename T, void(T::* callable)()>

void Start(T* object)

{

m_Thread = std::thread(Threading::HandleExceptions<DoStart<T, callable>>::Wrapper, object);

}

template<typename T, void(T::* callable)()>

static void DoStart(T* object);

protected:

~Thread()

{

ENSURE(!m_Thread.joinable());

}

std::thread m_Thread;

std::atomic<bool> m_Kill = false;

};

/**

* Worker thread: process the taskManager queues until killed.

class ThreadPool::WorkerThread : public Thread

{

public:

WorkerThread(ThreadPool::TaskManager::Impl& taskManager, WorkerAffinity affinity);

~WorkerThread();

StanUnsubmitted

Not Done

What's the point with only one value?

Stan: What's the point with only one value?

wraitiiAuthorUnsubmitted

Done

I had another value earlier... I guess I could make it a boolean again.

wraitii: I had another value earlier... I guess I could make it a boolean again.

/**

* Mark the worker for destruction. May need to be woken up.

void Stop();

std::atomic<bool> m_HasWork = false;

vladislavbelovUnsubmitted

Not Done

Should not be public.

vladislavbelov: Should not be public.

protected:

void RunUntilDeath();

WorkerAffinity m_Affinity;

ThreadPool::TaskManager::Impl& m_TaskManager;

};

/**

* PImpl-ed implementation of the threadpool's task manager.

* The normal priority queue is processed first, the low priority only if there are no higher-priority tasks

StanUnsubmitted

Not Done

Technically it's not the Xth thread pool, it's the xth thread in the thread pool :P

Stan: Technically it's not the Xth thread pool, it's the xth thread in the thread pool :P

wraitiiAuthorUnsubmitted

Done

The fact it's a thread name is obvious from context, and on linux I only have 16 characters.

wraitii: The fact it's a thread name is obvious from context, and on linux I only have 16 characters.

class ThreadPool::TaskManager::Impl

{

StanUnsubmitted

Not Done

Does this have any value? I mean literally anything could be going on in that thread and we already have the PROFILE2 macro used for the calling code.

Stan: Does this have any value? I mean literally anything could be going on in that thread and we…

wraitiiAuthorUnsubmitted

Done

literally crashes profiler2 to call without registering the thread.

wraitii: literally crashes profiler2 to call without registering the thread.

friend class TaskManager;

friend class WorkerThread;

public:

Impl(TaskManager& backref);

~Impl()

{

ClearQueue();

for (WorkerThread& worker : m_Workers)

worker.Stop();

m_ConditionVariable.notify_all();

m_Workers.clear();

}

/**

* 2-phase init to avoid having to think too hard about the order of class members.

void SetupWorkers(size_t numberOfWorkers);

/**

* Push a task on the global queue.

* Takes ownership of @a task.

* May be called from any thread.

void PushTask(std::function<void()>&& task, WorkerAffinity affinity, Priority priority);

WorkerAffinity GetAffinity() const;

protected:

void ClearQueue();

// Back reference (keep this first).

TaskManager& m_TaskManager;

std::mutex m_GlobalMutex;

std::mutex m_GlobalLowPriorityMutex;

std::deque<QueueItem> m_GlobalQueue;

std::deque<QueueItem> m_GlobalLowPriorityQueue;

// All workers share a mutex & a condition variable.

std::mutex m_WorkersMutex;

std::condition_variable m_ConditionVariable;

vladislavbelovUnsubmitted

Not Done

Instead of duplication and a bit of performance loose you could have a condition variable in each thread.

vladislavbelov: Instead of duplication and a bit of performance loose you could have a condition variable in…

// Ideally this would be a vector, since it does get iterated, but that requires movable types.

std::list<WorkerThread> m_Workers;

vladislavbelovUnsubmitted

Not Done

Might be deque.

vladislavbelov: Might be `deque`.

// TODO: won't be necessary when m_Workers can be a vector

std::vector<std::atomic<bool>*> m_WorkersWorkToggle;

// Cached.

std::vector<WorkerAffinity> m_WorkerAffinities;

// Round-robin counter for GetWorker.

mutable size_t m_RoundRobinIdx = 0;

};

ThreadPool::TaskManager::TaskManager() : TaskManager(std::thread::hardware_concurrency() - 1)

{

}

ThreadPool::TaskManager::TaskManager(size_t numberOfWorkers)

{

m = std::make_unique<Impl>(*this);

numberOfWorkers = Clamp<size_t>(numberOfWorkers, MIN_THREADS, MAX_THREADS);

m->SetupWorkers(numberOfWorkers);

}

ThreadPool::TaskManager::~TaskManager() {}

ThreadPool::TaskManager::Impl::Impl(TaskManager& backref)

: m_TaskManager(backref)

{

}

void ThreadPool::TaskManager::Impl::SetupWorkers(size_t numberOfWorkers)

{

for (size_t i = 0; i < numberOfWorkers; ++i)

{

m_Workers.emplace_back(*this, WorkerAffinity(i));

vladislavbelovUnsubmitted

Not Done

Duplication, you need to select (from i) the worker affinity only once.

vladislavbelov: Duplication, you need to select (from `i`) the worker affinity only once.

m_WorkersWorkToggle.emplace_back(&m_Workers.back().m_HasWork);

m_WorkerAffinities.emplace_back(ThreadPool::TaskManager::CreateWorkerAffinity(i));

// Register the thread on Profiler2 and name it, for convenience.

// (This is called here because RunUntilDeath is templated and the static value won't work as expected then).

m_TaskManager.PushTask([]() {

// The profiler does better if the names are unique.

static std::atomic<int> n = 0;

std::string name = "ThreadPool #" + std::to_string(n++);

vladislavbelovUnsubmitted

Not Done

It doesn't make sense to have a shorter name, the full numberOfWorkers is pretty short already and more clear.

vladislavbelov: It doesn't make sense to have a shorter name, the full `numberOfWorkers` is pretty short…

StanUnsubmitted

Not Done

Maybe you could CStrIntern that first part ?

Stan: Maybe you could CStrIntern that first part ?

debug_SetThreadName(name.c_str());

g_Profiler2.RegisterCurrentThread(name);

StanUnsubmitted

Not Done

m->m_Executors.resize(n) ? same for the list and the condition variables?

Stan: m->m_Executors.resize(n) ? same for the list and the condition variables?

vladislavbelovUnsubmitted

Not Done

Why limited by a simple constant and not std::thread::hardware_concurrency() * constant?

vladislavbelov: Why limited by a simple constant and not `std::thread::hardware_concurrency() * constant`?

wraitiiAuthorUnsubmitted

Done

The use of a condition variable per-worker makes it potentially inefficient with too many workers.
I could switch to a single CV if worker didn't have their own queue, but the way I currently implement thread affinity makes use of that and it seemed potentially more efficient.

That being said, if I switch to the task manager holding all tasks, it could be changed.

wraitii: The use of a condition variable per-worker makes it potentially inefficient with too many…

vladislavbelovUnsubmitted

Not Done

TaskManager holds all tasks in its queue.

vladislavbelov: `TaskManager` holds all tasks in its queue.

}, WorkerAffinity(i), Priority::NORMAL).Wait();

}

void ThreadPool::TaskManager::ClearQueue() { m->ClearQueue(); }

void ThreadPool::TaskManager::Impl::ClearQueue()

{

std::lock_guard<std::mutex> lock(m_GlobalMutex);

m_GlobalQueue.clear();

}

{

std::lock_guard<std::mutex> lock(m_GlobalLowPriorityMutex);

m_GlobalLowPriorityQueue.clear();

}

for (std::atomic<bool>* hasWork : m_WorkersWorkToggle)

*hasWork = false;

}

size_t ThreadPool::TaskManager::GetNbOfWorkers() const

wraitiiAuthorUnsubmitted

Done

This becomes increasingly suboptimal as users have more cores available. I'm not sure it's a huge problem event at 64 cores though.

wraitii: This becomes increasingly suboptimal as users have more cores available. I'm not sure it's a…

{

return m->m_Workers.size();

}

wraitiiAuthorUnsubmitted

Done

This can register too many hooks, but that doesn't really matter since it doesn't re-create the threads.

wraitii: This can register too many hooks, but that doesn't really matter since it doesn't re-create the…

ThreadPool::WorkerAffinity ThreadPool::TaskManager::GetAffinity() const

{

return m->GetAffinity();

}

ThreadPool::WorkerAffinity ThreadPool::TaskManager::Impl::GetAffinity() const

{

if (m_RoundRobinIdx >= m_Workers.size())

m_RoundRobinIdx = 0;

return WorkerAffinity(m_RoundRobinIdx++);

}

const std::vector<ThreadPool::WorkerAffinity>& ThreadPool::TaskManager::WorkerAffinities() const

{

return m->m_WorkerAffinities;

}

void ThreadPool::TaskManager::DoPushTask(std::function<void()>&& task, WorkerAffinity affinity, Priority priority)

{

m->PushTask(std::move(task), affinity, priority);

}

void ThreadPool::TaskManager::Impl::PushTask(std::function<void()>&& task, WorkerAffinity affinity, Priority priority)

{

std::mutex& mutex = priority == Priority::NORMAL ? m_GlobalMutex : m_GlobalLowPriorityMutex;

std::deque<QueueItem>& queue = priority == Priority::NORMAL ? m_GlobalQueue : m_GlobalLowPriorityQueue;

{

std::lock_guard<std::mutex> lock(mutex);

queue.emplace_back(std::move(task), affinity);

}

if (affinity == NoAffinity)

{

for (std::atomic<bool>* hasWork : m_WorkersWorkToggle)

*hasWork = true;

m_ConditionVariable.notify_one();

}

else

{

*m_WorkersWorkToggle[affinity.m_Index] = true;

m_ConditionVariable.notify_all();

}

void ThreadPool::TaskManager::Initialise()

{

if (!g_TaskManager)

g_TaskManager = std::make_unique<TaskManager>();

}

ThreadPool::TaskManager& ThreadPool::TaskManager::Instance()

{

ENSURE(g_TaskManager);

return *g_TaskManager;

}

// Thread definition

ThreadPool::WorkerThread::WorkerThread(ThreadPool::TaskManager::Impl& taskManager, WorkerAffinity affinity)

: m_TaskManager(taskManager), m_Affinity(affinity)

{

Start<WorkerThread, &WorkerThread::RunUntilDeath>(this);

}

ThreadPool::WorkerThread::~WorkerThread()

{

m_Kill = true;

if (m_Thread.joinable())

m_Thread.join();

}

void ThreadPool::WorkerThread::Stop()

{

m_Kill = true;

}

void ThreadPool::WorkerThread::RunUntilDeath()

{

std::function<void()> task;

vladislavbelovUnsubmitted

Not Done

detach doesn't sound good here, it means you make the wait implicit.

vladislavbelov: `detach` doesn't sound good here, it means you make the wait implicit.

bool hasTask = false;

std::unique_lock<std::mutex> lock(m_TaskManager.m_WorkersMutex);

while (!m_Kill)

{

m_TaskManager.m_ConditionVariable.wait(lock, [this]() -> bool {

return m_HasWork || m_Kill;

});

lock.unlock();

if (m_Kill)

break;

// Reset to false, we'll set it to true if we actually found work.

// This makes us do one 'bonus' iteration at the end, but simplifies book-keeping.

m_HasWork = false;

hasTask = false;

// Fetch work from the global queues if necessary.

{

// Particularly critical section since we're locking the global queue.

std::unique_lock<std::mutex> globalLock(m_TaskManager.m_GlobalMutex);

for (std::deque<QueueItem>::const_iterator it = m_TaskManager.m_GlobalQueue.begin(); it != m_TaskManager.m_GlobalQueue.end(); ++it)

if (it->m_Affinity == TaskManager::NoAffinity || it->m_Affinity == m_Affinity)

{

StanUnsubmitted

Not Done

Maybe we could pass the name?

Stan: Maybe we could pass the name?

task = std::move(it->m_Task);

hasTask = true;

m_TaskManager.m_GlobalQueue.erase(it);

break;

}

vladislavbelovUnsubmitted

Not Done

That logic should be hidden inside TaskManager.

vladislavbelov: That logic should be hidden inside `TaskManager`.

}

if (!hasTask)

{

std::unique_lock<std::mutex> globalLock(m_TaskManager.m_GlobalLowPriorityMutex);

for (std::deque<QueueItem>::const_iterator it = m_TaskManager.m_GlobalLowPriorityQueue.begin(); it != m_TaskManager.m_GlobalLowPriorityQueue.end(); ++it)

StanUnsubmitted

Not Done

merge the m_Kill if?

Stan: merge the m_Kill if?

wraitiiAuthorUnsubmitted

Done

I'd rather not, since those are relatively unrelated blocks, and in a concurrent environment, it's plausible the first one would return false and the second true.

wraitii: I'd rather not, since those are relatively unrelated blocks, and in a concurrent environment…

if (it->m_Affinity == TaskManager::NoAffinity || it->m_Affinity == m_Affinity)

{

task = std::move(it->m_Task);

hasTask = true;

m_TaskManager.m_GlobalLowPriorityQueue.erase(it);

break;

}

vladislavbelovUnsubmitted

Not Done

Code duplication.

vladislavbelov: Code duplication.

}

if (!hasTask)

{

lock.lock();

continue;

}

m_HasWork = true;

task();

lock.lock();

}

// Defined here - needs access to derived types.

template<typename T, void(T::* callable)()>

void ThreadPool::Thread::DoStart(T* object)

{

std::invoke(callable, object);

}

StanUnsubmitted

Not Done

Shouldn't it be the responsibility of the thread to give the correct pointer ?

Stan: Shouldn't it be the responsibility of the thread to give the correct pointer ?

wraitiiAuthorUnsubmitted

Done

wraitii: ?

StanUnsubmitted

Not Done

Using an accesor or something

Stan: Using an accesor or something

wraitiiAuthorUnsubmitted

Done

No, this is a member of the executor - just a trick to make a temporary thread without showing it to the client code.

wraitii: No, this is a member of the executor - just a trick to make a temporary thread without showing…

StanUnsubmitted

Done

Is there a way to do it in a non tricky way?

Stan: Is there a way to do it in a non tricky way?

wraitiiAuthorUnsubmitted

Done

It's not that tricky, & no without forcing the client code to store a reference to the ThreadExecutor, which is ugly IMO.

wraitii: It's not that tricky, & no without forcing the client code to store a reference to the…

StanUnsubmitted

Not Done

The conditions are a bit hard to read since they are overlapping a bit. Maybe there is another way to test ?

Stan: The conditions are a bit hard to read since they are overlapping a bit. Maybe there is another…