Something went wrong. Try again.
OpenMW is an open implementation of the Morrowind engine
Something went wrong. Try again.
3.2 kB · 145 lines
C++
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146#include "workqueue.hpp"
#include <components/debug/debuglog.hpp>
#include <numeric>
namespace SceneUtil{
void WorkItem::waitTillDone() { if (mDone) return;
std::unique_lock<std::mutex> lock(mMutex); while (!mDone) { mCondition.wait(lock); } }
void WorkItem::signalDone() { { std::unique_lock<std::mutex> lock(mMutex); mDone = true; } mCondition.notify_all(); }
bool WorkItem::isDone() const { return mDone; }
WorkQueue::WorkQueue(std::size_t workerThreads) : mIsReleased(false) { start(workerThreads); }
WorkQueue::~WorkQueue() { stop(); }
void WorkQueue::start(std::size_t workerThreads) { { const std::lock_guard lock(mMutex); mIsReleased = false; } while (mThreads.size() < workerThreads) mThreads.emplace_back(std::make_unique<WorkThread>(*this)); }
void WorkQueue::stop() { { std::unique_lock<std::mutex> lock(mMutex); while (!mQueue.empty()) mQueue.pop_back(); mIsReleased = true; mCondition.notify_all(); }
mThreads.clear(); }
void WorkQueue::addWorkItem(std::shared_ptr<WorkItem> item, bool front) { if (item->isDone()) { Log(Debug::Error) << "Error: trying to add a work item that is already completed"; return; }
std::unique_lock<std::mutex> lock(mMutex); if (front) mQueue.push_front(std::move(item)); else mQueue.push_back(std::move(item)); mCondition.notify_one(); }
std::shared_ptr<WorkItem> WorkQueue::removeWorkItem() { std::unique_lock<std::mutex> lock(mMutex); while (mQueue.empty() && !mIsReleased) { mCondition.wait(lock); } if (!mQueue.empty()) { std::shared_ptr<WorkItem> item = std::move(mQueue.front()); mQueue.pop_front(); return item; } return nullptr; }
size_t WorkQueue::getNumItems() const { std::unique_lock<std::mutex> lock(mMutex); return mQueue.size(); }
size_t WorkQueue::getNumActiveThreads() const { return std::accumulate( mThreads.begin(), mThreads.end(), 0u, [](auto r, const auto& t) { return r + t->isActive(); }); }
WorkThread::WorkThread(WorkQueue& workQueue) : mWorkQueue(&workQueue) , mActive(false) , mThread([this] { run(); }) { }
WorkThread::~WorkThread() { mThread.join(); }
void WorkThread::run() { while (true) { std::shared_ptr<WorkItem> item = mWorkQueue->removeWorkItem(); if (!item) return; mActive = true; item->doWork(); item->signalDone(); mActive = false; } }
bool WorkThread::isActive() const { return mActive; }
}