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

#include "Simulator.h"
#include "parex_protocol/NodeProtocol.h"
#include "parex_protocol/SimResult.h"
#include "parex_protocol/SimTask.h"
#include "parex_server/QHostAnyAddress.h"

#include <QtGlobal>
#include <QPushButton>
#include <QString>
#include <QTcpSocket>
#include <QThread>
#include <cassert>
#include <iostream>

using namespace std;

namespace SimPT_Parex {

const int WorkerNode::g_broadcast_port = 45678;

WorkerNode::WorkerNode(Simulator* simulator, bool verbose, QObject* parent)
        : QTcpServer(parent), m_simulator(simulator), m_simulator_thread(new QThread(this)),
          m_advertiser(g_broadcast_port, verbose), m_verbose(verbose), m_protocol(nullptr),
          m_current_task(nullptr), m_last_address(QHostAddress::Null), m_queue(), m_results()
{
        assert(simulator && "simulator cannot be a nullptr");

        qRegisterMetaType<SimTask>("SimTask");
        qRegisterMetaType<SimResult>("SimResult");
        qRegisterMetaType<std::string>("std::string");

        //listen(QHostAddress::AnyIPv4);
        listen(QHostAnyAddress());

        connect(this, SIGNAL(newConnection()), this, SLOT(ConnectionReceived()));

        StartAdvertiser();

        m_simulator->moveToThread(m_simulator_thread);
        connect(m_simulator, SIGNAL(TaskSolved(const SimResult&)),
                this, SLOT(FinishedWork(const SimResult&)));
        connect(this, SIGNAL(NewTask(const SimTask&)),
                m_simulator, SLOT(SolveTask(const SimTask&)));
        m_simulator_thread->start();
}

WorkerNode::~WorkerNode()
{
        close();
        m_simulator->deleteLater();
        m_simulator_thread->quit();
        m_simulator_thread->wait();
}

void WorkerNode::ConnectionReceived()
{
        if (m_verbose) {
                cout << "Connection received" << endl;
        }

        QTcpSocket* socket = nextPendingConnection();
        if (!socket) {
                return;
        }

        if (!m_protocol) {
                m_protocol = new NodeProtocol(socket, this);

                connect(m_protocol, SIGNAL(Ended()), m_protocol, SLOT(deleteLater()));
                connect(m_protocol, SIGNAL(Error(const std::string&)),
                        this, SLOT(DisplayError(const std::string&)));

                // Connects for solving a SimTask and returning the result
                connect(m_protocol, SIGNAL(TaskReceived(const SimTask*)),
                        this, SLOT(StartedWork(const SimTask*)));
                connect(m_protocol, SIGNAL(StopTask()), this, SLOT(StopTask()));
                connect(m_protocol, SIGNAL(Delete(const std::string&)),
                        this, SLOT(Delete(const std::string&)));

                // this slot is to another thread and will be executed after
                // the current simulation finishes, if one is running
                connect(m_protocol, SIGNAL(Delete(const std::string&)),
                        m_simulator, SLOT(Delete(const std::string&)));

                connect(m_protocol, SIGNAL(SuccessfullySent()), this, SLOT(ResultSent()));

                connect(socket, SIGNAL(error(QAbstractSocket::SocketError)),
                        this, SLOT(HandleError(QAbstractSocket::SocketError)));

                if (socket->peerAddress() == m_last_address) {
                        // Check for not-confirmed simulations
                        if (!m_results.empty())
                                m_protocol->SendSimResult(m_results.front());
                } else {
                        while (!m_queue.empty()) {
                                delete m_queue.front();
                                m_queue.pop();
                        }

                        while (!m_results.empty()) {
                                m_results.pop();
                        }
                }
                m_last_address = socket->peerAddress();
                m_advertiser.Stop();
        } else {
                cerr << "Already connected, ignoring new connection." << endl;
                socket->close();
                delete socket;
        }
}

void WorkerNode::Delete(const std::string& name)
{
        //a delete on the current exploration
        if (m_current_task && name == m_current_task->GetExploration()) {
                m_simulator->StopTask();
        }
}

void WorkerNode::DisplayError(const std::string& error) const
{
        if (m_verbose) {
                cerr << "WorkerNode Error: " << error << endl;
        }
}

void WorkerNode::FinishedWork(const SimResult& result)
{
        delete m_current_task;
        m_current_task = nullptr;

        m_results.push(result);
        if (m_protocol && m_protocol->IsConnected()) {
                m_protocol->SendSimResult(result);
        } else {
                if (m_verbose)
                        cout << "Disconnected when finished task" << endl;
        }

        if (!m_queue.empty()) {
                if (m_verbose)
                        cout << "More work in the queue, starting that now" << endl;
                m_current_task = m_queue.front();
                m_queue.pop();
        } else {
                if (m_verbose)
                        cout << "Finished job, waiting for more work" << endl;
        }
}

void WorkerNode::HandleError(QAbstractSocket::SocketError )
{
        delete m_protocol;
        m_protocol = nullptr;
        StartAdvertiser();
}

void WorkerNode::ResultSent()
{
        m_results.pop();
        if (!m_results.empty() && m_protocol && m_protocol->IsConnected()) {
                m_protocol->SendSimResult(m_results.front());
        }
}

void WorkerNode::StartAdvertiser()
{
        if (!m_results.empty()) {
                m_advertiser.Start(serverPort(),
                        m_results.front().GetExplorationName(), m_results.front().GetTaskId());
        } else if (m_current_task) {
                m_advertiser.Start(serverPort(),
                        m_current_task->GetExploration(), m_current_task->GetId());
        } else {
                m_advertiser.Start(serverPort());
        }
}

void WorkerNode::StartedWork(const SimTask* task)
{
        if (!m_current_task){
                m_current_task = task;
                emit NewTask(*task);
        } else {
                if (m_verbose) {
                        cerr << "WARNING: Received work while still working, queuing the work" << endl;
                }
                m_queue.push(task);
        }
}

void WorkerNode::StopTask()
{
        m_simulator->StopTask();
}

} // namespace