Added support for grouping InsightWorker tasks
This commit is contained in:
@@ -12,49 +12,76 @@ namespace Bloom
|
||||
using Bloom::Targets::TargetState;
|
||||
|
||||
void InsightWorker::startup() {
|
||||
QObject::connect(
|
||||
InsightSignals::instance(),
|
||||
&InsightSignals::taskQueued,
|
||||
this,
|
||||
&InsightWorker::executeTasks
|
||||
);
|
||||
auto* insightSignals = InsightSignals::instance();
|
||||
|
||||
QObject::connect(insightSignals, &InsightSignals::taskQueued, this, &InsightWorker::executeTasks);
|
||||
QObject::connect(insightSignals, &InsightSignals::taskProcessed, this, &InsightWorker::executeTasks);
|
||||
|
||||
Logger::debug("InsightWorker" + std::to_string(this->id) + " thread ready");
|
||||
emit this->ready();
|
||||
}
|
||||
|
||||
void InsightWorker::queueTask(InsightWorkerTask* task) {
|
||||
static std::atomic<QueuedTaskId> lastQueuedTaskId = 0;
|
||||
task->moveToThread(nullptr);
|
||||
|
||||
{
|
||||
const auto taskQueueLock = InsightWorker::queuedTasks.acquireLock();
|
||||
InsightWorker::queuedTasks.getValue().push(task);
|
||||
const auto taskQueueLock = InsightWorker::queuedTasksById.acquireLock();
|
||||
InsightWorker::queuedTasksById.getValue().insert(std::pair(++(lastQueuedTaskId), task));
|
||||
}
|
||||
|
||||
emit InsightSignals::instance()->taskQueued();
|
||||
}
|
||||
|
||||
std::optional<InsightWorkerTask*> InsightWorker::getQueuedTask() {
|
||||
auto taskQueueLock = InsightWorker::queuedTasks.acquireLock();
|
||||
auto& queuedTasks = InsightWorker::queuedTasks.getValue();
|
||||
|
||||
if (!queuedTasks.empty()) {
|
||||
auto* task = queuedTasks.front();
|
||||
queuedTasks.pop();
|
||||
return task;
|
||||
}
|
||||
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
void InsightWorker::executeTasks() {
|
||||
static const auto getQueuedTask = [] () -> std::optional<InsightWorkerTask*> {
|
||||
const auto taskQueueLock = InsightWorker::queuedTasksById.acquireLock();
|
||||
auto& queuedTasks = InsightWorker::queuedTasksById.getValue();
|
||||
|
||||
if (!queuedTasks.empty()) {
|
||||
const auto taskGroupsLock = InsightWorker::taskGroupsInExecution.acquireLock();
|
||||
auto& taskGroupsInExecution = InsightWorker::taskGroupsInExecution.getValue();
|
||||
|
||||
const auto canExecuteTask = [&taskGroupsInExecution] (InsightWorkerTask* task) {
|
||||
for (const auto taskGroup : task->getTaskGroups()) {
|
||||
if (taskGroupsInExecution.contains(taskGroup)) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
};
|
||||
|
||||
for (auto [queuedTaskId, task] : queuedTasks) {
|
||||
if (canExecuteTask(task)) {
|
||||
const auto taskGroups = task->getTaskGroups();
|
||||
taskGroupsInExecution.insert(taskGroups.begin(), taskGroups.end());
|
||||
queuedTasks.erase(queuedTaskId);
|
||||
return task;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return std::nullopt;
|
||||
};
|
||||
|
||||
auto queuedTask = std::optional<InsightWorkerTask*>();
|
||||
|
||||
while ((queuedTask = InsightWorker::getQueuedTask()).has_value()) {
|
||||
while ((queuedTask = getQueuedTask())) {
|
||||
auto* task = queuedTask.value();
|
||||
task->moveToThread(this->thread());
|
||||
task->setParent(this);
|
||||
task->execute(this->targetControllerConsole);
|
||||
|
||||
{
|
||||
const auto taskGroupsLock = InsightWorker::taskGroupsInExecution.acquireLock();
|
||||
auto& taskGroupsInExecution = InsightWorker::taskGroupsInExecution.getValue();
|
||||
|
||||
for (const auto& taskGroup : task->getTaskGroups()) {
|
||||
taskGroupsInExecution.erase(taskGroup);
|
||||
}
|
||||
}
|
||||
|
||||
task->deleteLater();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
#include <atomic>
|
||||
#include <QtCore>
|
||||
#include <queue>
|
||||
#include <map>
|
||||
|
||||
#include "Tasks/InsightWorkerTask.hpp"
|
||||
|
||||
@@ -33,12 +34,15 @@ namespace Bloom
|
||||
void ready();
|
||||
|
||||
private:
|
||||
using QueuedTaskId = std::uint64_t;
|
||||
static_assert(std::atomic<QueuedTaskId>::is_always_lock_free);
|
||||
|
||||
static inline std::atomic<std::uint8_t> lastWorkerId = 0;
|
||||
static inline SyncSafe<std::queue<InsightWorkerTask*>> queuedTasks = {};
|
||||
static inline SyncSafe<std::map<QueuedTaskId, InsightWorkerTask*>> queuedTasksById = {};
|
||||
static inline SyncSafe<TaskGroups> taskGroupsInExecution = {};
|
||||
|
||||
TargetController::TargetControllerConsole targetControllerConsole = TargetController::TargetControllerConsole();
|
||||
|
||||
static std::optional<InsightWorkerTask*> getQueuedTask();
|
||||
void executeTasks();
|
||||
};
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
#include <QObject>
|
||||
#include <QString>
|
||||
|
||||
#include "TaskGroup.hpp"
|
||||
#include "src/TargetController/TargetControllerConsole.hpp"
|
||||
|
||||
namespace Bloom
|
||||
@@ -24,6 +25,10 @@ namespace Bloom
|
||||
|
||||
InsightWorkerTask(): QObject(nullptr) {};
|
||||
|
||||
virtual TaskGroups getTaskGroups() const {
|
||||
return TaskGroups();
|
||||
};
|
||||
|
||||
void execute(TargetController::TargetControllerConsole& targetControllerConsole);
|
||||
|
||||
signals:
|
||||
|
||||
12
src/Insight/InsightWorker/Tasks/TaskGroup.hpp
Normal file
12
src/Insight/InsightWorker/Tasks/TaskGroup.hpp
Normal file
@@ -0,0 +1,12 @@
|
||||
#pragma once
|
||||
|
||||
#include <cstdint>
|
||||
#include <set>
|
||||
|
||||
namespace Bloom
|
||||
{
|
||||
enum class TaskGroup: std::uint16_t
|
||||
{};
|
||||
|
||||
using TaskGroups = std::set<TaskGroup>;
|
||||
}
|
||||
Reference in New Issue
Block a user