ExplorationManager.cpp

Go to the documentation of this file.

/*
 * Copyright 2011-2016 Universiteit Antwerpen
 *
 * Licensed under the EUPL, Version 1.1 or  as soon they will be approved by
 * the European Commission - subsequent versions of the EUPL (the "Licence");
 * You may not use this work except in compliance with the Licence.
 * You may obtain a copy of the Licence at: http://ec.europa.eu/idabc/eupl5
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the Licence is distributed on an "AS IS" basis,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the Licence for the specific language governing
 * permissions and limitations under the Licence.
 */
#include "ExplorationManager.h"

#include "WorkerPool.h"
#include "WorkerRepresentative.h"
#include "parex_protocol/Exploration.h"
#include "parex_protocol/ExplorationProgress.h"
#include "parex_protocol/ExplorationTask.h"
#include "parex_protocol/SimTask.h"
#include "util/misc/StringUtils.h"

#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/xml_parser.hpp>
#include <algorithm>
#include <cassert>
#include <iostream>
#include <sstream>

using namespace std;
using boost::property_tree::ptree;
using namespace boost::property_tree::xml_parser;

namespace SimPT_Parex {

ExplorationManager::ExplorationManager()
{
        Read_Backup();

        //Make sure the WorkerPool is listening to connecting nodes by calling globalInstance()
        connect(WorkerPool::globalInstance(), SIGNAL(NewWorkerAvailable()),
                this, SLOT(NewWorkerAvailable()));
        connect(WorkerPool::globalInstance(), SIGNAL(WorkerReconnected(WorkerRepresentative*)),
                this, SLOT(WorkerReconnected(WorkerRepresentative*)));
}

ExplorationManager::~ExplorationManager()
{
}

ExplorationProgress* ExplorationManager::GetExploration(const std::string& name,
                                const std::deque<ExplorationProgress*>& explorations)
{
        auto matching_iter = std::find_if(explorations.begin(), explorations.end(),
                [&name] (ExplorationProgress* e) { return name == e->GetExploration().GetName(); });

        if (matching_iter != explorations.end()) {
                return (*matching_iter);
        } else {
                return nullptr;
        }
}

void ExplorationManager::RemoveExploration(const std::string& name,
                                std::deque<ExplorationProgress*>& explorations)
{
        auto matching_iter = std::find_if(explorations.begin(), explorations.end(),
                [&name] (ExplorationProgress* e) { return name == e->GetExploration().GetName(); });

        if (matching_iter != explorations.end()) {
                explorations.erase(matching_iter);
        }
}

std::vector<std::string> ExplorationManager::GetExplorationNames()
{
        std::vector<std::string> names;

        for (auto progress : m_running_explorations) {
                names.push_back(progress->GetExploration().GetName());
        }
        for (auto progress : m_done_explorations) {
                names.push_back(progress->GetExploration().GetName());
        }

        return names;

}

bool ExplorationManager::WorkAvailable()
{
        if (m_running_explorations.empty()){
                return false;
        }

        for (auto progress : m_running_explorations) {
                if (progress->GetTaskCount(TaskState::Waiting) != 0)
                        return true;
        }

        return false;
}

void ExplorationManager::RegisterExploration(const Exploration* exploration)
{
        if (exploration->GetNumberOfTasks() == 0) {
                delete exploration;
                return;
        }

        ExplorationProgress* progress = new ExplorationProgress(exploration);
        m_running_explorations.push_back(progress);
        connect(progress, SIGNAL(Updated()), this, SIGNAL(Updated()));
        connect(progress, SIGNAL(Updated()), this, SLOT(BackUp()));
        while (WorkerPool::globalInstance()->WorkerAvailable() && WorkAvailable()) {
                NewWorkerAvailable();
        }

        BackUp();
 }

void ExplorationManager::DeleteExploration(const std::string& name)
{
        auto exploration = GetExploration(name, m_running_explorations);

        if (exploration) {
                RemoveExploration(name, m_running_explorations);
                delete exploration;
        } else {
                exploration = GetExploration(name, m_done_explorations);
                if (exploration) {
                        RemoveExploration(name, m_done_explorations);
                        delete exploration;
                }
        }
        WorkerPool::globalInstance()->Delete(name);
        BackUp();
        emit Updated();
 }

const ExplorationProgress* ExplorationManager::GetExplorationProgress(const std::string& explorationName)
{
        auto progress = GetExploration(explorationName, m_running_explorations);
        if (progress == nullptr)
                progress = GetExploration(explorationName, m_done_explorations);
        if (progress == nullptr)
                return nullptr;

        return progress;
}

void ExplorationManager::NewWorkerAvailable()
{
        if (m_running_explorations.empty()) {
                return;
        }

        bool done = false;
        ExplorationProgress* progress = nullptr;

        unsigned int i = 0;
        while (!done && i < m_running_explorations.size()) {
                //scheduling is FIFO
                progress = m_running_explorations.at(i);

                if (progress->GetTaskCount(TaskState::Waiting) > 0) {
                        done = true;
                } else {
                        progress = nullptr;
                }
                ++i;
        }

        if (!progress) {
                return;
        }

        auto task = progress->NextTask();
        assert(task && "ExplorationProgress has waiting tasks, but NextTask returns a nullptr");

        auto worker = WorkerPool::globalInstance()->getProcess();
        if (!worker) {
                //Maybe someone else was faster in retrieving a worker.
                progress->GiveBack(task);
                return;
        }

        connect(worker, SIGNAL(FinishedWork(const SimResult&)),
                progress, SLOT(HandleResult(const SimResult&)), Qt::UniqueConnection);
        connect(worker, SIGNAL(FinishedWork(const SimResult&)),
                this, SLOT(WorkerFinished()), Qt::UniqueConnection);
        connect(worker, SIGNAL(FinishedWork(const SimResult&)),
                this, SLOT(BackUp()), Qt::UniqueConnection);

        connect(worker, SIGNAL(Disconnected()),
                this, SLOT(WorkerDisconnected()), Qt::UniqueConnection);

        m_running_tasks[worker] = task;
        worker->DoWork(*task);

        BackUp();
}

void ExplorationManager::WorkerFinished()
{
        WorkerRepresentative* worker = dynamic_cast<WorkerRepresentative*>(QObject::sender());
        assert(worker && "QObject::sender() is not a WorkerRepresentative*");

        auto progress = GetExploration(worker->GetExplName(), m_running_explorations);
        if (!progress)
                return;

        if (progress->IsFinished()) {
                m_done_explorations.push_back(progress);
                RemoveExploration(worker->GetExplName(), m_running_explorations);
        }

        delete m_running_tasks[worker];
        m_running_tasks.erase(worker);
}

void ExplorationManager::WorkerDisconnected()
{
        WorkerRepresentative* worker = dynamic_cast<WorkerRepresentative*>(QObject::sender());
        assert(worker && "QObject::sender() is not a WorkerRepresentative*");

        SimTask* task = m_running_tasks[worker];

        if (task == nullptr) {
                m_running_tasks.erase(worker);
                return;
        }

        auto progress = GetExploration(task->GetExploration(), m_running_explorations);

        if (!progress)
                return;

        progress->GiveBack(task);
        m_running_tasks.erase(worker);
}

void ExplorationManager::WorkerReconnected(WorkerRepresentative* )
{
        // TODO Choose what to do when a worker reconnects.
        // What is the best choice: stop all other tasks and reassign and let the same node continue?
        // Or just give the reconnected node a new task and let the other one finish the task the disconnected node was doing?

//      SimTask *task = new SimTask(rep->GetTaskId(), "", rep->GetExplName());
//
//      /* Check if the exploration is sent by this Manager */
//      auto matching_iter = std::find_if(m_running_explorations.begin(), m_running_explorations.end(),
//              [task] (ExplorationProgress* e) {
//                      return task->GetExploration() == e->GetExploration().GetName();
//              });
//
//      if (matching_iter == m_running_explorations.end())
//              return;
//
//      for (auto tasks : m_running_tasks) {
//              if (tasks.second->GetExploration() == task->GetExploration() &&
//                      tasks.second->GetId() == task->GetId()) {
//                      tasks.first->StopTask();
//                      return;
//              }
//      }
//
//      m_running_tasks[rep] = task;
//
//      //make sure we don't have a double connection
//      for (auto progress : m_running_explorations) {
//              disconnect(rep, SIGNAL(FinishedWork(const SimResult&)), progress, SLOT(Finished(const SimResult&)));
//      }
//      for (auto progress : m_done_explorations) {
//              disconnect(rep, SIGNAL(FinishedWork(const SimResult&)), progress, SLOT(Finished(const SimResult&)));
//      }
//
//      auto progress = *matching_iter;
//
//      // TODO Hmmm, so the task would've been set to Waiting-state...
//      // In the meanwhile, another worker might have gotten this task
//      // What was the meaning of this?
//      //progress->ResendCancel(task->GetId());
//
//      connect(rep, SIGNAL(FinishedWork(const SimResult&)), progress, SLOT(Finished(const SimResult&)));
//      connect(rep, SIGNAL(FinishedWork(const SimResult&)), this, SLOT(WorkerFinished()));
//      connect(rep, SIGNAL(FinishedWork(const SimResult&)), this, SLOT(BackUp()));
//
//      disconnect(rep, SIGNAL(Disconnected()), this, SLOT(WorkerDisconnected()));
//      connect(rep, SIGNAL(Disconnected()), this, SLOT(WorkerDisconnected()));

        NewWorkerAvailable();
}

void ExplorationManager::BackUp()
{
        ptree writer;

        writer.put("RunningExplorationsNr", m_running_explorations.size());
        int i = 0;
        for (auto progress : m_running_explorations) {
                writer.put_child("RunningExplorations" + to_string(i), progress->ToPtree());
                ++i;
        }

        i = 0;
        writer.put("DoneExplorationsNr", m_done_explorations.size());
        for (auto progress : m_done_explorations) {
                writer.put_child("DoneExplorations" + to_string(i), progress->ToPtree());
                ++i;
        }

        write_xml("exploration_backup.xml", writer);
}

void ExplorationManager::Read_Backup()
{
        try {
                ptree reader;
                read_xml("exploration_backup.xml", reader);

                for (int i = 0; i < reader.get<int>("RunningExplorationsNr"); i++) {
                        ExplorationProgress* progress
                                = new ExplorationProgress(
                                        reader.get_child("RunningExplorations" + to_string(i)));
                        connect(progress, SIGNAL(Updated()), this, SIGNAL(Updated()));
                        m_running_explorations.push_back(progress);
                }

                for (int i = 0; i < reader.get<int>("DoneExplorationsNr"); i++) {
                        ExplorationProgress* progress
                                = new ExplorationProgress(
                                        reader.get_child("DoneExplorations" + to_string(i)));
                        connect(progress, SIGNAL(Updated()), this, SIGNAL(Updated()));
                        m_done_explorations.push_back(progress);
                }
        }
        catch (std::exception& e) {

        }
        catch (...) {
                return;
        }
}

void ExplorationManager::StopTask(const std::string& name, int id)
{
        auto progress = GetExploration(name, m_running_explorations);
        if (!progress)
                return;

        auto state = progress->GetTask(id).GetState();
        if (state == TaskState::Running) {
                for (auto task : m_running_tasks) {
                        if (task.second->GetExploration() == name && task.second->GetId() == id) {
                                task.first->StopTask();
                                return;
                        }
                }
        } else if (state == TaskState::Waiting) {
                progress->CancelWaitingTask(id);
        }
}

void ExplorationManager::RestartTask(const std::string& name, int id)
{
        auto progress = GetExploration(name, m_running_explorations);
        if (!progress) {
                progress = GetExploration(name, m_done_explorations);
                if (!progress)
                        return;

                m_running_explorations.push_back(progress);
                RemoveExploration(name, m_done_explorations);
        }

        if (progress->GetTask(id).GetState() == TaskState::Cancelled) {
                progress->ResendCancelledTask(id);
        }

        while (WorkerPool::globalInstance()->WorkerAvailable() && WorkAvailable()) {
                NewWorkerAvailable();
        }
}

} // namespace