From 11ea7fcb71a5b61fa4918a0977da8eb86c0b6380 Mon Sep 17 00:00:00 2001 From: Bennett Date: Tue, 3 May 2022 17:57:22 -0700 Subject: [PATCH] Initial threadpool rewrite/cleanup --- app/render/rendermanager.cpp | 38 +++----- app/threading/threadpool.cpp | 175 +++++++++++------------------------ app/threading/threadpool.h | 81 ++++++---------- 3 files changed, 91 insertions(+), 203 deletions(-) diff --git a/app/render/rendermanager.cpp b/app/render/rendermanager.cpp index ac5d297de..a20033217 100644 --- a/app/render/rendermanager.cpp +++ b/app/render/rendermanager.cpp @@ -40,8 +40,7 @@ RenderManager* RenderManager::instance_ = nullptr; const int RenderManager::kDecoderMaximumInactivity = 10000; RenderManager::RenderManager(QObject *parent) : - ThreadPool(QThread::IdlePriority, 0, parent), - backend_(kOpenGL) + ThreadPool(0, parent), backend_(kOpenGL) { Renderer* graphics_renderer = nullptr; @@ -160,14 +159,7 @@ RenderTicketPtr RenderManager::RenderFrame(ViewerOutput *viewer, ColorManager* c ticket->setProperty("cache", cache->GetCacheDirectory()); } - if (ticket->thread() != this->thread()) { - ticket->moveToThread(this->thread()); - } - - // Queue appending the ticket and running the next job on our thread to make this function thread-safe - QMetaObject::invokeMethod(this, "AddTicket", Qt::AutoConnection, - OLIVE_NS_ARG(RenderTicketPtr, ticket), - Q_ARG(bool, prioritize)); + AddTicket(ticket); return ticket; } @@ -189,14 +181,7 @@ RenderTicketPtr RenderManager::RenderAudio(ViewerOutput* viewer, const TimeRange ticket->setProperty("enablewaveforms", generate_waveforms); ticket->setProperty("aparam", QVariant::fromValue(params)); - if (ticket->thread() != this->thread()) { - ticket->moveToThread(this->thread()); - } - - // Queue appending the ticket and running the next job on our thread to make this function thread-safe - QMetaObject::invokeMethod(this, "AddTicket", Qt::AutoConnection, - OLIVE_NS_ARG(RenderTicketPtr, ticket), - Q_ARG(bool, prioritize)); + AddTicket(ticket); return ticket; } @@ -211,20 +196,21 @@ RenderTicketPtr RenderManager::SaveFrameToCache(FrameHashCache *cache, FramePtr ticket->setProperty("hash", hash); ticket->setProperty("type", kTypeVideoDownload); - if (ticket->thread() != this->thread()) { - ticket->moveToThread(this->thread()); - } - - // Queue appending the ticket and running the next job on our thread to make this function thread-safe - QMetaObject::invokeMethod(this, "AddTicket", Qt::AutoConnection, - OLIVE_NS_ARG(RenderTicketPtr, ticket), - Q_ARG(bool, prioritize)); + AddTicket(ticket); return ticket; } void RenderManager::RunTicket(RenderTicketPtr ticket) const { + // Setup the ticket for ::Process + ticket->Start(); + + if (ticket->IsCancelled()) { + ticket->Finish(); + return; + } + RenderProcessor::Process(ticket, context_, decoder_cache_, shader_cache_, default_shader_); } diff --git a/app/threading/threadpool.cpp b/app/threading/threadpool.cpp index cef8500c0..96a5f1106 100644 --- a/app/threading/threadpool.cpp +++ b/app/threading/threadpool.cpp @@ -22,140 +22,71 @@ namespace olive { -ThreadPool::ThreadPool(QThread::Priority priority, int threads, QObject *parent) : - QObject(parent) +ThreadPool::ThreadPool(unsigned threads, QObject *parent) + : QObject(parent) { - all_threads_.resize(threads ? threads : QThread::idealThreadCount()); + if (threads == 0) { + threads = std::thread::hardware_concurrency(); + } - // Create threads - for (int i=0; i lock(task_mutex_); - // Append to list of available threads - available_threads_.push_back(t); + if (priority == RenderTicketPriority::kHigh) { + tasks_.emplace_front(std::move(ticket)); + } else { + tasks_.emplace_back(std::move(ticket)); + } - // Connect done signal - connect(t, &ThreadPoolThread::Done, this, &ThreadPool::ThreadDone); + cond_.notify_one(); +} - // Start the thread at the given priority - t->start(priority); +bool ThreadPool::RemoveTicket(RenderTicketPtr ticket) +{ + std::lock_guard lock(task_mutex_); + + const auto it = std::find(tasks_.begin(), tasks_.end(), ticket); + if (it == tasks_.end()) { + return false; + } + + tasks_.erase(it); + return true; +} + +void ThreadPool::thread_exec() { + while (true) { + TaskType task; + + { + std::unique_lock lock(task_mutex_); + cond_.wait(lock, [this]{ return this->end_threadp_ || !this->tasks_.empty(); }); + + if (this->end_threadp_ && this->tasks_.empty()) { + break; + } + + task = std::move(tasks_.front()); + tasks_.pop_front(); + } + + RunTicket(task); } } ThreadPool::~ThreadPool() { - foreach (ThreadPoolThread* thread, all_threads_) { - thread->Cancel(); - thread->wait(); - delete thread; + end_threadp_ = true; + cond_.notify_all(); + + for (auto &e : worker_threads_) { + e.join(); } } -bool ThreadPool::RemoveTicket(RenderTicketPtr ticket) -{ - auto it = std::find(ticket_queue_.begin(), ticket_queue_.end(), ticket); - if (it == ticket_queue_.end()) { - return false; - } - - ticket_queue_.erase(it); - return true; -} - -void ThreadPool::AddTicket(RenderTicketPtr ticket, bool prioritize) -{ - if (prioritize) { - ticket_queue_.push_front(ticket); - } else { - ticket_queue_.push_back(ticket); - } - - RunNext(); -} - -void ThreadPool::RunNext() -{ - while (!ticket_queue_.empty() && !available_threads_.empty()) { - // Run function - RenderTicketPtr ticket = ticket_queue_.front(); - ticket_queue_.pop_front(); - - ticket->Start(); - - if (ticket->IsCancelled()) { - // Finish without doing any more - ticket->Finish(); - } else { - ThreadPoolThread* thread = available_threads_.front(); - available_threads_.pop_front(); - - // Move ticket to other thread so event processing can occur there - ticket->moveToThread(thread); - - // Run the ticket in the thread, which actually just calls our virtual function RunTicket - thread->RunTicket(ticket); - } - } -} - -void ThreadPool::ThreadDone() -{ - ThreadPoolThread* thread = static_cast(sender()); - - available_threads_.push_back(thread); - - RunNext(); -} - -ThreadPoolThread::ThreadPoolThread(ThreadPool *parent) -{ - pool_ = parent; - - // Ensures mutex is definitely locked by the time the thread is running - mutex_.lock(); -} - -ThreadPoolThread::~ThreadPoolThread() -{ - mutex_.unlock(); -} - -void ThreadPoolThread::RunTicket(RenderTicketPtr ticket) -{ - mutex_.lock(); - ticket_ = ticket; - wait_cond_.wakeAll(); - mutex_.unlock(); -} - -void ThreadPoolThread::run() -{ - while (true) { - wait_cond_.wait(&mutex_); - - if (ticket_) { - pool_->RunTicket(ticket_); - - // Move back to calling thread (hacky?) - ticket_->moveToThread(this->thread()); - - ticket_ = nullptr; - } - - if (IsCancelled()) { - break; - } else { - emit Done(); - } - } -} - -void ThreadPoolThread::CancelEvent() -{ - wait_cond_.wakeAll(); -} - } diff --git a/app/threading/threadpool.h b/app/threading/threadpool.h index ee1e29b61..e56fef52a 100644 --- a/app/threading/threadpool.h +++ b/app/threading/threadpool.h @@ -21,75 +21,46 @@ #ifndef THREADPOOL_H #define THREADPOOL_H -#include - -#include "common/cancelableobject.h" #include "threading/threadticket.h" +#include +#include +#include +#include +#include + namespace olive { -class ThreadPoolThread; +enum class RenderTicketPriority { kHigh = 0, kNormal }; class ThreadPool : public QObject { Q_OBJECT public: - ThreadPool(QThread::Priority priority = QThread::InheritPriority, int threads = 0, QObject* parent = nullptr); + using TaskType = RenderTicketPtr; + ThreadPool(unsigned threads, QObject *parent); + + ThreadPool(const ThreadPool &) = delete; + ThreadPool(ThreadPool &&) = delete; + ThreadPool & operator=(const ThreadPool&) = delete; + ThreadPool & operator=(ThreadPool &&) = delete; + + virtual void RunTicket(RenderTicketPtr ticket) const = 0; + void AddTicket(RenderTicketPtr ticket, RenderTicketPriority priority = RenderTicketPriority::kNormal); + bool RemoveTicket(RenderTicketPtr ticket); virtual ~ThreadPool() override; - RenderTicketPtr Queue(); - - virtual void RunTicket(RenderTicketPtr ticket) const = 0; - - bool RemoveTicket(RenderTicketPtr ticket); - -public slots: - void AddTicket(olive::RenderTicketPtr ticket, bool prioritize = false); - private: - void RunNext(); - - QVector all_threads_; - - std::list available_threads_; - - std::list ticket_queue_; - -private slots: - void ThreadDone(); + void thread_exec(); + std::vector worker_threads_; + std::deque tasks_; + std::mutex task_mutex_; + std::condition_variable cond_; + std::atomic_bool end_threadp_{false}; }; -class ThreadPoolThread : public QThread, public CancelableObject -{ - Q_OBJECT -public: - ThreadPoolThread(ThreadPool* parent); +} // namespace olive - virtual ~ThreadPoolThread() override; - - void RunTicket(RenderTicketPtr ticket); - -protected: - virtual void run() override; - - virtual void CancelEvent() override; - -signals: - void Done(); - -private: - ThreadPool* pool_; - - RenderTicketPtr ticket_; - - QMutex mutex_; - - QWaitCondition wait_cond_; - -}; - -} - -#endif // THREADPOOL_H +#endif // THREADPOOL_H