thijn/SqMod
1
0
Fork 0
mirror of https://github.com/VCMP-SqMod/SqMod.git synced 2025-06-16 07:07:13 +02:00
Code Releases Activity

Allow CPR to use the thread pool.

Browse Source
This commit is contained in:
Sandu Liviu Catalin
2021-03-28 00:19:09 +02:00
parent 6a0b7f246e
commit cd55e51d62
6 changed files with 492 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
module/Core/ThreadPool.cpp
View File

@ -10,10 +10,18 @@ namespace SqMod {
// ------------------------------------------------------------------------------------------------
ThreadPool ThreadPool::s_Inst;
// ------------------------------------------------------------------------------------------------
void ProcessThreads()
{
ThreadPool::Get().Process();
}
// ------------------------------------------------------------------------------------------------
ThreadPool::ThreadPool() noexcept
: m_Running(false)
, m_Pending()
, m_Mutex()
, m_CV()
, m_Queue()
, m_Finished()
, m_Threads()
{
@ -31,6 +39,8 @@ ThreadPool::~ThreadPool()
t.join(); // Will block until work is finished!
}
}
// Clear all thread instances
m_Threads.clear();
}
// ------------------------------------------------------------------------------------------------
@ -46,21 +56,12 @@ bool ThreadPool::Initialize(uint32_t count)
{
count = MAX_WORKER_THREADS; // Hard coded worker limit
}
// See if any of the threads are active
for (auto & t : m_Threads)
{
if (t.joinable())
{
return false; // Something is not right!
}
}
// Make sure the threads don't stop after creation
m_Running = true;
// Create the specified amount of worker threads
for (uint32_t i = 0; i < count; ++i)
{
// Pass the container index to the worker thread so it knows to find itself
m_Threads.emplace_back(&ThreadPool::WorkerProc, this, i);
m_Threads.emplace_back(&ThreadPool::WorkerProc, this);
}
// Thread pool initialized
return m_Running;
@ -76,10 +77,15 @@ void ThreadPool::Terminate(bool SQ_UNUSED_ARG(shutdown))
}
// Tell the threads to stop
m_Running = false;
// Enqueue dummy items to wake the threads and allow them to stop
for (size_t n = 0; n < m_Threads.size(); ++n)
{
m_Pending.enqueue(Item());
std::lock_guard< std::mutex > lg(m_Mutex);
// Enqueue dummy items to wake the threads and allow them to stop
for (size_t n = 0; n < m_Threads.size(); ++n)
{
m_Queue.push(Item());
}
// Allow threads to process the dummy items and stop
m_CV.notify_all();
}
// Attempt to join the threads
for (auto & t : m_Threads)
@ -89,6 +95,8 @@ void ThreadPool::Terminate(bool SQ_UNUSED_ARG(shutdown))
t.join(); // Will block until work is finished!
}
}
// Clear all thread instances
m_Threads.clear();
// Retrieve each item individually and process it
for (Item item; m_Finished.try_dequeue(item);)
{
@ -124,7 +132,7 @@ void ThreadPool::Process()
}
// ------------------------------------------------------------------------------------------------
void ThreadPool::WorkerProc(uint32_t SQ_UNUSED_ARG(idx))
void ThreadPool::WorkerProc()
{
// Whether this item wants to try again
bool retry = false;
@ -147,7 +155,17 @@ void ThreadPool::WorkerProc(uint32_t SQ_UNUSED_ARG(idx))
// Attempt to get an item from the queue
if (!retry)
{
m_Pending.wait_dequeue(item);
// Acquire a lock on the mutex
std::unique_lock< std::mutex > lock(m_Mutex);
// Wait until there are items in the queue
while (m_Queue.empty())
{
m_CV.wait(lock);
}
// Retrieve top item the queue
item = std::move(m_Queue.front());
// Remove it from the queue
m_Queue.pop();
}
// Do we have to stop?
if (!m_Running)

60
module/Core/ThreadPool.hpp
View File

@ -5,7 +5,6 @@
// ------------------------------------------------------------------------------------------------
#include <concurrentqueue.h>
#include <blockingconcurrentqueue.h>
// ------------------------------------------------------------------------------------------------
#include <queue>
@ -13,6 +12,7 @@
#include <vector>
#include <atomic>
#include <thread>
#include <condition_variable>
// ------------------------------------------------------------------------------------------------
namespace SqMod {
@ -40,6 +40,11 @@ struct ThreadPoolItem
*/
ThreadPoolItem(ThreadPoolItem && o) = delete;
/* --------------------------------------------------------------------------------------------
* Destructor.
*/
virtual ~ThreadPoolItem() = default;
/* --------------------------------------------------------------------------------------------
* Copy assignment operator. (disabled)
*/
@ -54,19 +59,18 @@ struct ThreadPoolItem
* Invoked in worker thread by the thread pool after obtaining the task from the queue.
* Must return true to indicate that the task can be performed. False indicates failure.
*/
virtual bool OnPrepare() { return true; }
SQMOD_NODISCARD virtual bool OnPrepare() { return true; }
/* --------------------------------------------------------------------------------------------
* Called in worker by the thread pool to performed by the associated tasks.
* Will be called continuously while the returned value is true. While false means it finished.
*/
virtual bool OnProcess() = 0;
SQMOD_NODISCARD virtual bool OnProcess() { return false; };
/* --------------------------------------------------------------------------------------------
* Invoked in main thread by the thread pool after the task was completed.
* If true is returned, the task is added back to the queue to be processed again.
*/
virtual bool OnCompleted() { return false; }
virtual void OnCompleted() { }
/* --------------------------------------------------------------------------------------------
* Called in worker by the thread pool to let the task know that it will be aborted.
@ -75,20 +79,6 @@ struct ThreadPoolItem
virtual void OnAborted(bool SQ_UNUSED_ARG(retry)) { }
};
/* ------------------------------------------------------------------------------------------------
* Common implementation of a basic item.
*/
struct BasicThreadPoolItem
{
// --------------------------------------------------------------------------------------------
LightObj mCallback{};
/* --------------------------------------------------------------------------------------------
* Default constructor.
*/
BasicThreadPoolItem() noexcept = default;
};
/* ------------------------------------------------------------------------------------------------
* Internal thread pool used to reduce stuttering from the plug-in whenever necessary and/or possible.
*/
@ -115,23 +105,25 @@ private:
using Pool = std::vector< std::thread >; // Worker container.
using Item = std::unique_ptr< ThreadPoolItem >; // Owning pointer of an item.
// --------------------------------------------------------------------------------------------
using Pending = moodycamel::BlockingConcurrentQueue< Item >; // Pending items.
using Finished = moodycamel::ConcurrentQueue< Item >; // Finished items.
// --------------------------------------------------------------------------------------------
std::atomic_bool m_Running; // Whether the threads are allowed to run.
std::atomic_bool m_Running; // Whether the threads are allowed to run.
// --------------------------------------------------------------------------------------------
Pending m_Pending; // Blocking concurrent queue of pending items.
Finished m_Finished; // Non-blocking concurrent queue of finished items.
std::mutex m_Mutex;
std::condition_variable m_CV;
std::queue< Item > m_Queue;
// --------------------------------------------------------------------------------------------
Pool m_Threads; // Pool of worker threads.
Finished m_Finished; // Non-blocking concurrent queue of finished items.
// --------------------------------------------------------------------------------------------
Pool m_Threads; // Pool of worker threads.
private:
/* --------------------------------------------------------------------------------------------
* Internal function used to process tasks.
*/
void WorkerProc(uint32_t idx);
void WorkerProc();
public:
@ -188,7 +180,14 @@ public:
// Only queue if worker threads exist
if (!m_Threads.empty())
{
m_Pending.enqueue(Item(item));
// Acquire a lock on the mutex
std::unique_lock< std::mutex > lock(m_Mutex);
// Push the item in the queue
m_Queue.push(Item(item));
// Release the mutex before notifying
lock.unlock();
// Notify one thread that there's work
m_CV.notify_one();
}
else
{
@ -204,6 +203,15 @@ public:
item->OnCompleted();
}
}
/* --------------------------------------------------------------------------------------------
* Retrieve the number of worker threads.
*/
SQMOD_NODISCARD size_t GetThreadCount()
{
return m_Threads.size();
}
};
} // Namespace:: SqMod