/***
Olive - Non-Linear Video Editor
Copyright (C) 2019 Olive Team
This program 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 3 of the License, or
(at your option) any later version.
This program 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 this program. If not, see .
***/
#include "taskmanager.h"
#include
#include
OLIVE_NAMESPACE_ENTER
TaskManager* TaskManager::instance_ = nullptr;
TaskManager::TaskManager() :
active_thread_count_(0)
{
// Initialize threads to run tasks on
threads_.resize(QThread::idealThreadCount());
for (int i=0;istart(QThread::IdlePriority);
threads_.replace(i, {t, false});
}
}
TaskManager::~TaskManager()
{
// First send the signal to all tasks to start cancelling
foreach (const TaskContainer& task_info, tasks_) {
if (task_info.status == kWorking) {
task_info.task->Cancel();
}
}
// Next, signal each thread to quit as its next event in the queue
foreach (const ThreadContainer& tc, threads_) {
tc.thread->quit();
}
// Wait for each thread's event queue to finish
foreach (const ThreadContainer& tc, threads_) {
tc.thread->wait();
// This is technically unnecessary since each QThread is a child of this object, but we may as well
delete tc.thread;
}
// Finally delete all task objects (they shouldn't have been deleted by TaskSucceeded() or TaskFailed() because our
// event queue shouldn't be active by this point
foreach (const TaskContainer& task_info, tasks_) {
delete task_info.task;
}
}
void TaskManager::CreateInstance()
{
instance_ = new TaskManager();
}
void TaskManager::DestroyInstance()
{
delete instance_;
instance_ = nullptr;
}
TaskManager *TaskManager::instance()
{
return instance_;
}
int TaskManager::GetTaskCount() const
{
return tasks_.size();
}
Task *TaskManager::GetFirstTask() const
{
return tasks_.first().task;
}
void TaskManager::AddTask(Task* t)
{
// Connect Task's status signal to the Callback
connect(t, &Task::Succeeded, this, &TaskManager::TaskSucceeded, Qt::QueuedConnection);
connect(t, &Task::Failed, this, &TaskManager::TaskFailed, Qt::QueuedConnection);
connect(t, &Task::Finished, this, &TaskManager::TaskFinished, Qt::QueuedConnection);
// Add the Task to the queue
tasks_.append({t, kWaiting});
// Emit signal that a Task was added
emit TaskAdded(t);
emit TaskListChanged();
// Scan through queue and start any Tasks that can (including this one)
StartNextWaiting();
}
void TaskManager::StartNextWaiting()
{
// If there are no tasks in the queue, there is nothing to be done
if (tasks_.isEmpty()) {
return;
}
// If all threads are occupied, nothing to be done
if (active_thread_count_ == threads_.size()) {
return;
}
// Create a list of tasks that are waiting
QList waiting_tasks;
foreach (const TaskContainer& task_info, tasks_) {
if (task_info.status == kWaiting) {
waiting_tasks.append(task_info.task);
}
}
// No tasks waiting to start
if (waiting_tasks.isEmpty()) {
return;
}
// For any inactive threads,
for (int i=0;imoveToThread(threads_.at(i).thread);
threads_[i].active = true;
active_thread_count_++;
SetTaskStatus(task, kWorking);
QMetaObject::invokeMethod(task,
"Start",
Qt::QueuedConnection);
if (active_thread_count_ == threads_.size() || waiting_tasks.isEmpty()) {
break;
}
}
}
}
void TaskManager::DeleteTask(Task *t)
{
if (GetTaskStatus(t) == kWorking) {
// Send a signal to the task to cancel, it will likely continue to cancel in the background after it's removed
t->Cancel();
}
// Remove instances of Task from queue
for (int i=0;iRemoved();
emit TaskListChanged();
if (GetTaskStatus(t) != kWorking) {
// If the task isn't doing anything, we can simply delete it
t->deleteLater();
}
}
void TaskManager::TaskFinished()
{
Task* task_sender = static_cast(sender());
// Set this thread's active value to false
for (int i=0;ithread()) {
threads_[i].active = false;
}
}
// See if we can delete this task
if (GetTaskStatus(task_sender) == kFinished) {
DeleteTask(task_sender);
} else if (GetTaskStatus(task_sender) == kError) {
// If this task has already been deleted, we'll free its memory now
bool was_deleted = true;
for (int i=0;ideleteLater();
}
}
// Decrement the active thread count
active_thread_count_--;
// Signal that the task has finished
emit TaskListChanged();
// Start any tasks that could start now
StartNextWaiting();
}
TaskManager::TaskStatus TaskManager::GetTaskStatus(Task *t)
{
foreach (const TaskContainer& container, tasks_) {
if (container.task == t) {
return container.status;
}
}
return kError;
}
void TaskManager::SetTaskStatus(Task *t, TaskStatus status)
{
for (int i=0;i(sender()), kFinished);
}
void TaskManager::TaskFailed()
{
SetTaskStatus(static_cast(sender()), kError);
}
OLIVE_NAMESPACE_EXIT