From 44d81f6b00d74fe3dcf12de76b09da315c0942d5 Mon Sep 17 00:00:00 2001
From: itsmattkc <34096995+itsmattkc@users.noreply.github.com>
Date: Mon, 30 May 2022 22:11:06 -0700
Subject: [PATCH] render: dedicate thread to audio production

Prevents audio from getting stalled behind slow video jobs
---
 app/threading/threadpool.cpp | 55 +++++++++++++++++++++++-------------
 app/threading/threadpool.h   |  9 +++++-
 app/widget/viewer/viewer.cpp |  9 +++---
 3 files changed, 49 insertions(+), 24 deletions(-)

diff --git a/app/threading/threadpool.cpp b/app/threading/threadpool.cpp
index b048f3519..8b33d72db 100644
--- a/app/threading/threadpool.cpp
+++ b/app/threading/threadpool.cpp
@@ -29,52 +29,67 @@ ThreadPool::ThreadPool(unsigned threads, QObject *parent) :
     threads = std::thread::hardware_concurrency();
   }
 
+  available_count_ = threads;
   for (unsigned i = 0; i < threads; i += 1) {
-    worker_threads_.emplace_back(std::bind(&ThreadPool::thread_exec, this));
+    worker_threads_.emplace_back(std::bind(&ThreadPool::thread_exec, this, &tasks_, &task_mutex_, &cond_));
   }
+
+  // Make single reserved thread for high priority tasks (usually audio) so they don't get stuck
+  // behind a lot of slow tasks
+  high_thread_ = std::thread(std::thread(std::bind(&ThreadPool::thread_exec, this, &high_tasks_, &high_mutex_, &high_cond_)));
 }
 
 void ThreadPool::AddTicket(RenderTicketPtr ticket, RenderTicketPriority priority)
 {
-  std::lock_guard<std::mutex> lock(task_mutex_);
-
   if (priority == RenderTicketPriority::kHigh) {
-    tasks_.emplace_front(std::move(ticket));
+    std::lock_guard<std::mutex> lock(high_mutex_);
+    high_tasks_.emplace_back(std::move(ticket));
+    high_cond_.notify_one();
   } else {
+    std::lock_guard<std::mutex> lock(task_mutex_);
     tasks_.emplace_back(std::move(ticket));
+    cond_.notify_one();
   }
-
-  cond_.notify_one();
 }
 
 bool ThreadPool::RemoveTicket(RenderTicketPtr ticket)
 {
-  std::lock_guard<std::mutex> lock(task_mutex_);
-
-  const auto it = std::find(tasks_.begin(), tasks_.end(), ticket);
-  if (it == tasks_.end()) {
-    return false;
+  {
+    std::lock_guard<std::mutex> lock(task_mutex_);
+    const auto it = std::find(tasks_.begin(), tasks_.end(), ticket);
+    if (it != tasks_.end()) {
+      tasks_.erase(it);
+      return true;
+    }
   }
 
-  tasks_.erase(it);
-  return true;
+  {
+    std::lock_guard<std::mutex> lock(high_mutex_);
+    const auto it = std::find(high_tasks_.begin(), high_tasks_.end(), ticket);
+    if (it != high_tasks_.end()) {
+      high_tasks_.erase(it);
+      return true;
+    }
+  }
+
+  return false;
 }
 
-void ThreadPool::thread_exec()
+void ThreadPool::thread_exec(std::deque<TaskType> *queue, std::mutex *mutex, std::condition_variable *cond)
 {
   while (true) {
     TaskType task;
 
     {
-      std::unique_lock<std::mutex> lock(task_mutex_);
-      cond_.wait(lock, [this]{ return this->end_threadp_ || !this->tasks_.empty(); });
+      std::unique_lock<std::mutex> lock(*mutex);
+      cond->wait(lock, [this, queue]{ return this->end_threadp_ || !queue->empty(); });
 
-      if (this->end_threadp_ && this->tasks_.empty()) {
+      if (this->end_threadp_ && queue->empty()) {
         break;
       }
 
-      task = std::move(tasks_.front());
-      tasks_.pop_front();
+      task = std::move(queue->front());
+      queue->pop_front();
     }
 
     RunTicket(task);
@@ -85,10 +100,12 @@ ThreadPool::~ThreadPool()
 {
   end_threadp_ = true;
   cond_.notify_all();
+  high_cond_.notify_all();
 
   for (auto &e : worker_threads_) {
     e.join();
   }
+  high_thread_.join();
 }
 
 }
diff --git a/app/threading/threadpool.h b/app/threading/threadpool.h
index 3c3204bb1..3c1143f55 100644
--- a/app/threading/threadpool.h
+++ b/app/threading/threadpool.h
@@ -49,13 +49,20 @@ public:
   virtual ~ThreadPool() override;
 
 private:
-  void thread_exec();
+  void thread_exec(std::deque<TaskType> *queue, std::mutex *mutex, std::condition_variable *cond);
 
   std::vector<std::thread> worker_threads_;
   std::deque<TaskType> tasks_;
   std::mutex task_mutex_;
   std::condition_variable cond_;
+
+  std::thread high_thread_;
+  std::deque<TaskType> high_tasks_;
+  std::mutex high_mutex_;
+  std::condition_variable high_cond_;
+
   std::atomic_bool end_threadp_{false};
+  std::atomic_int available_count_;
 
 };
 
diff --git a/app/widget/viewer/viewer.cpp b/app/widget/viewer/viewer.cpp
index f807b2d19..35eb765d0 100644
--- a/app/widget/viewer/viewer.cpp
+++ b/app/widget/viewer/viewer.cpp
@@ -1492,10 +1492,11 @@ void ViewerWidget::PlaybackTimerUpdate()
   }
 
   if (IsPlaying()) {
-    int count = 0;
-    for (int i=display_widget_->queue()->size(); i<DeterminePlaybackQueueSize(); i++) {
-      RequestNextFrameForQueue();
-      count++;
+    while (queue_watchers_.size() < DeterminePlaybackQueueSize()) {
+      if (!RequestNextFrameForQueue()) {
+        // Prevent infinite loop
+        break;
+      }
     }
   }