WorkerPool.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 "WorkerPool.h"

#include "QHostAnyAddress.h"
#include "WorkerRepresentative.h"

#include <QByteArray>
#include <QCoreApplication>
#include <QDebug>
#include <QProcess>

namespace {
        //The interval on which the number of nodes is checked (in msec)
        const int C_WAIT_TIME = 30 * 1000;
}

namespace SimPT_Parex {

WorkerPool::WorkerPool()
        : m_socket(new QUdpSocket()), m_tcpSocket(0), m_mapper(new QSignalMapper(this)),
          m_release_mapper(new QSignalMapper(this)), m_initialized(false),
          m_initialize_timer(nullptr), m_min_nodes(0), m_check_timer(new QTimer(this))
{
        Initialize();

}

void WorkerPool::Initialize()
{
        if (!m_socket->bind(QHostAnyAddress(), 45678, QUdpSocket::ShareAddress)) {
                std::cerr << "WARNING: couldn't bind to the discovery port";
                if (!m_initialize_timer) {
                        m_initialize_timer = new QTimer();
                        connect(m_initialize_timer, SIGNAL(timeout()), this, SLOT(Initialize()));
                        m_initialize_timer->start(1000);
                }
                return;
        }

        m_socket->joinMulticastGroup(QHostAddress("239.13.13.13"));
        connect(m_socket, SIGNAL(readyRead()), this, SLOT(handleDatagrams()));
        connect(m_mapper, SIGNAL(mapped(QObject *)), this, SLOT(MakeProcessAvailable(QObject *)));
        connect(m_release_mapper, SIGNAL(mapped(QObject *)), this, SLOT(ReleaseWorker(QObject *)));

        delete m_initialize_timer;
        m_initialized = true;
        connect(QCoreApplication::instance(), SIGNAL(aboutToQuit()), this, SLOT(deleteLater()));
        connect(m_check_timer, SIGNAL(timeout()), this, SLOT(PeriodicCheck()));
        m_check_timer->setSingleShot(false);
        m_check_timer->start(C_WAIT_TIME);
}

WorkerPool::~WorkerPool()
{
        m_socket->close();
        delete m_socket;
}

bool contains(const std::list<QString>& l, QString s)
{
        for (QString q : l) {
                if (s == q)
                        return true;
        }
        return false;
}

void WorkerPool::handleDatagrams()
{
        //We have to process all pending datagrams or the readyRead() signal will not be emitted again
        while (m_socket->hasPendingDatagrams()) {
                QByteArray datagram;
                datagram.resize(m_socket->pendingDatagramSize());

                //We don't know where the node is located, so check the sender address of the UDP packet
                QHostAddress *senderAddr = new QHostAddress();

                m_socket->readDatagram(datagram.data(), datagram.size(), senderAddr);

                QString s(datagram);
                int port = 0;
                int taskId = -1;
                std::string taskName = "";
                QString portString = "";

                const int FIND_SERVER_MESSAGE = 0;
                const int FIND_SERVER_WHILE_WORKING_MESSAGE = 1;
                const int BAD_MESSAGE = -1;

                int messageType;
                try {
                        if (s.startsWith("Job Application")) {
                                QString msg("Job Application");
                                s.remove(0, msg.length() + 1); //remove the server message + " "
                                portString = s;
                                port = portString.toInt();
                                messageType = FIND_SERVER_MESSAGE;
                        } else if (s.startsWith("Server Discovery")) {
                                QString msg("Server Discovery");
                                s.remove(0, msg.length() + 1);
                                int index = s.indexOf(" ");
                                QString idString = s.left(index + 1);   //index is 0based, left is not
                                taskId = idString.toInt();
                                s.remove(0, index + 1); //index is 0based, remove is not
                                index = s.indexOf(" ");
                                portString = s.left(index + 1); //index is 0based, left is not
                                port = portString.toInt();
                                s.remove(0, index + 1); //index is 0based, remove is not
                                taskName = s.toStdString();
                                messageType = FIND_SERVER_WHILE_WORKING_MESSAGE;
                        }
                }
                catch (std::exception &) {
                        messageType = BAD_MESSAGE;
                }
                if (!port) {
                        messageType = BAD_MESSAGE;
                }

                switch (messageType) {
                case FIND_SERVER_MESSAGE:
                        if (senderAddr && port) {
                                QString id = senderAddr->toString() + ":" + s;
                                if (!contains(m_connected_list, id)) {
                                        m_connected_list.push_back(id);

                                        WorkerRepresentative *rep = new WorkerRepresentative(*senderAddr, port);

                                        connect(rep, SIGNAL(ReadyToWork()), m_mapper, SLOT(map()));
                                        m_mapper->setMapping(rep, rep);

                                        connect(rep, SIGNAL(Disconnected()), m_release_mapper, SLOT(map()));
                                        m_release_mapper->setMapping(rep, rep);

                                        rep->Setup();
                                }
                        }
                        break;
                case FIND_SERVER_WHILE_WORKING_MESSAGE:
                        if (senderAddr && port) {
                                QString id = senderAddr->toString() + ":" + portString;
                                if (!contains(m_connected_list, id)) {
                                        m_connected_list.push_back(id);

                                        WorkerRepresentative *rep = new WorkerRepresentative(*senderAddr, port, taskId,
                                                taskName);

                                        connect(rep, SIGNAL(ReadyToWork()), m_mapper, SLOT(map()));
                                        m_mapper->setMapping(rep, rep);

                                        connect(rep, SIGNAL(Disconnected()), m_release_mapper, SLOT(map()));
                                        m_release_mapper->setMapping(rep, rep);

                                        rep->Setup();

                                        emit WorkerReconnected(rep);
                                }
                        }
                        break;
                default:
                        break;
                }

                delete senderAddr;
        }
}

WorkerPool *WorkerPool::globalInstance()
{
        static WorkerPool *pool = new WorkerPool();
        if (!pool) {
                //if the workerpool was destructed, make sure we create a new one
                pool = new WorkerPool();
        }
        return pool;
}

WorkerRepresentative *WorkerPool::getProcess()
{
        if (m_available_list.empty()) {
                return 0;
        }
        WorkerRepresentative* rep = m_available_list.front();
        m_available_list.pop_front();
        m_busy_list.push_back(rep);
        return rep;
}

void WorkerPool::MakeProcessAvailable(QObject *rep)
{
        MakeProcessAvailable(dynamic_cast<WorkerRepresentative*>(rep));
}

void WorkerPool::MakeProcessAvailable(WorkerRepresentative *rep)
{
        if (!rep) {
                return;
        }
        m_busy_list.remove(rep);
        m_available_list.push_back(rep);

        emit NewWorkerAvailable();
}

void WorkerPool::ReleaseWorker(QObject *rep)
{
        ReleaseWorker(dynamic_cast<WorkerRepresentative*>(rep));
}

void WorkerPool::ReleaseWorker(WorkerRepresentative *rep)
{
        if (!rep)
                return;
        m_busy_list.remove(rep);
        m_available_list.remove(rep);
        QString id = rep->GetSenderAddress()->toString() + ":" + QString::number(rep->GetSenderPort());
        m_connected_list.remove(id);
        rep->deleteLater();
}

bool WorkerPool::WorkerAvailable()
{
        return !m_available_list.empty();
}

void WorkerPool::StartWorker()
{
        QString nodePath = QCoreApplication::applicationFilePath();
        nodePath = nodePath.left(nodePath.lastIndexOf('/') + 1);
        nodePath += "parex_node";
        QProcess node;
        QStringList arguments;
        arguments << "-q";
        node.startDetached(nodePath, arguments);
}

void WorkerPool::StartWorkers()
{
        int neededWorkers = m_min_nodes - m_connected_list.size();
        for (int i = 0; i < neededWorkers; i++) {
                StartWorker();
        }
}

void WorkerPool::PeriodicCheck()
{
        StartWorkers();
        if (WorkerAvailable()) {
                emit NewWorkerAvailable();
        }
}

void WorkerPool::SetMinNumWorkers(int min_nodes)
{
        m_min_nodes = min_nodes;
}

void WorkerPool::Delete(std::string name)
{
        for (auto worker : m_available_list) {
                if (worker) {
                        worker->Delete(name);
                } else {
                        std::cout << "PROBLEM" << std::endl;
                }
        }
        for (auto worker : m_busy_list) {
                if (worker) {
                        worker->Delete(name);
                } else {
                        std::cout << "PROBLEM" << std::endl;
                }
        }
}

} // namespace