TaskDispatcher.cpp

#include "TaskDispatcher.h"

#include "QtConcurrent"

typedef std::unique_lock<std::mutex> ULock;

namespace nemo_interface {

TaskDispatcher::TaskDispatcher() : _running(false) {}

TaskDispatcher::~TaskDispatcher() { stop(); }

std::future<QVariant> TaskDispatcher::dispatch(std::unique_ptr<Task> c) {
  ULock lk1(this->_mutex);

  this->_taskQueue.push_back(std::move(c));
  std::promise<QVariant> promise;
  auto future = promise.get_future();
  this->_promiseQueue.push_back(std::move(promise));

  if (!this->_running) {
    lk1.unlock();
    this->start();
  }

  return future;
}

std::future<QVariant> TaskDispatcher::dispatchNext(std::unique_ptr<Task> c) {
  ULock lk1(this->_mutex);

  this->_taskQueue.push_front(std::move(c));
  std::promise<QVariant> promise;
  auto future = promise.get_future();
  this->_promiseQueue.push_front(std::move(promise));

  if (!this->_running) {
    lk1.unlock();
    this->start();
  }

  return future;
}

void TaskDispatcher::clear() {
  ULock lk(this->_mutex);
  this->_taskQueue.clear();
  this->_promiseQueue.clear();
}

void TaskDispatcher::start() {
  ULock lk1(this->_mutex);
  if (!_running) {
    this->_running = true;
    lk1.unlock();
    _future = QtConcurrent::run([this]() mutable { return this->run(); });
  }
}

void TaskDispatcher::stop() {
  ULock lk1(this->_mutex);
  if (_running) {
    this->_running = false;
    lk1.unlock();
    this->_future.waitForFinished();
  }
}

void TaskDispatcher::reset() {
  clear();
  stop();
}

bool TaskDispatcher::isInterruptionRequested() {
  ULock lk1(this->_mutex);
  return !_running;
}

bool TaskDispatcher::isRunning() {
  ULock lk1(this->_mutex);
  return _running;
}

std::size_t TaskDispatcher::pendingTasks() {
  ULock lk1(this->_mutex);
  return this->_taskQueue.size() + (_running ? 1 : 0);
}

void TaskDispatcher::run() {
  while (true) {
    ULock lk1(this->_mutex);

    if (this->_taskQueue.size() > 0 && this->_running) {
      Q_ASSERT(this->_taskQueue.size() == this->_promiseQueue.size());
      // pop task and promise
      auto pTask = std::move(this->_taskQueue.front());
      auto promise = std::move(this->_promiseQueue.front());
      this->_taskQueue.pop_front();
      this->_promiseQueue.pop_front();
      lk1.unlock();

      // exec task
      promise.set_value(pTask->exec());
    } else {
      this->_running = false;
      lk1.unlock();
      break;
    }
  }
}

} // namespace nemo_interface