task: massively simplified task system
Made various changes and fixes to the task system: - Tasks are built around QtConcurrent rather than QThread. Reduces code complexity significantly. - Task error reporting is now streamlined in both TaskManager and TaskDialog. - Moved ProjectImport/Save/LoadManager to the app/task folder
This commit is contained in:
itsmattkc
+29
-177
@@ -27,45 +27,22 @@ OLIVE_NAMESPACE_ENTER
|
||||
|
||||
TaskManager* TaskManager::instance_ = nullptr;
|
||||
|
||||
TaskManager::TaskManager() :
|
||||
active_thread_count_(0)
|
||||
TaskManager::TaskManager()
|
||||
{
|
||||
// Initialize threads to run tasks on
|
||||
threads_.resize(QThread::idealThreadCount());
|
||||
|
||||
for (int i=0;i<threads_.size();i++) {
|
||||
QThread* t = new QThread(this);
|
||||
t->start(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();
|
||||
}
|
||||
thread_pool_.clear();
|
||||
|
||||
foreach (Task* t, tasks_) {
|
||||
t->Cancel();
|
||||
}
|
||||
|
||||
// Next, signal each thread to quit as its next event in the queue
|
||||
foreach (const ThreadContainer& tc, threads_) {
|
||||
tc.thread->quit();
|
||||
}
|
||||
thread_pool_.waitForDone();
|
||||
|
||||
// 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;
|
||||
foreach (Task* t, tasks_) {
|
||||
t->deleteLater();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -92,171 +69,46 @@ int TaskManager::GetTaskCount() const
|
||||
|
||||
Task *TaskManager::GetFirstTask() const
|
||||
{
|
||||
return tasks_.first().task;
|
||||
return tasks_.begin().value();
|
||||
}
|
||||
|
||||
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);
|
||||
// Create a watcher for signalling
|
||||
QFutureWatcher<bool>* watcher = new QFutureWatcher<bool>();
|
||||
connect(watcher, &QFutureWatcher<bool>::finished, this, &TaskManager::TaskFinished);
|
||||
|
||||
// Add the Task to the queue
|
||||
tasks_.append({t, kWaiting});
|
||||
tasks_.insert(watcher, t);
|
||||
|
||||
// Run task concurrently
|
||||
watcher->setFuture(QtConcurrent::run(t, &Task::Run));
|
||||
|
||||
// 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<Task*> 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;i<threads_.size();i++) {
|
||||
if (!threads_.at(i).active) {
|
||||
// This thread is inactive and needs a new Task
|
||||
Task* task = waiting_tasks.takeFirst();
|
||||
|
||||
task->moveToThread(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;i<tasks_.size();i++) {
|
||||
if (tasks_.at(i).task == t) {
|
||||
tasks_.removeAt(i);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
emit t->Removed();
|
||||
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<Task*>(sender());
|
||||
QFutureWatcher<bool>* watcher = static_cast<QFutureWatcher<bool>*>(sender());
|
||||
Task* t = tasks_.value(watcher);
|
||||
|
||||
// Set this thread's active value to false
|
||||
for (int i=0;i<threads_.size();i++) {
|
||||
if (threads_.at(i).thread == task_sender->thread()) {
|
||||
threads_[i].active = false;
|
||||
}
|
||||
tasks_.remove(watcher);
|
||||
|
||||
if (watcher->result()) {
|
||||
// Task completed successfully
|
||||
emit TaskRemoved(t);
|
||||
t->deleteLater();
|
||||
} else {
|
||||
// Task failed, keep it so the user can see the error message
|
||||
emit TaskFailed(t);
|
||||
failed_tasks_.append(t);
|
||||
}
|
||||
|
||||
// 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;
|
||||
watcher->deleteLater();
|
||||
|
||||
for (int i=0;i<tasks_.size();i++) {
|
||||
if (tasks_.at(i).task == task_sender) {
|
||||
was_deleted = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (was_deleted) {
|
||||
task_sender->deleteLater();
|
||||
}
|
||||
}
|
||||
|
||||
// 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<tasks_.size();i++) {
|
||||
TaskContainer& cont = tasks_[i];
|
||||
|
||||
if (cont.task == t) {
|
||||
cont.status = status;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void TaskManager::TaskSucceeded()
|
||||
{
|
||||
SetTaskStatus(static_cast<Task*>(sender()), kFinished);
|
||||
}
|
||||
|
||||
void TaskManager::TaskFailed()
|
||||
{
|
||||
SetTaskStatus(static_cast<Task*>(sender()), kError);
|
||||
}
|
||||
|
||||
OLIVE_NAMESPACE_EXIT
|
||||
|
||||
Reference in New Issue
Block a user