AnalysisRunner.cpp

Go to the documentation of this file.

#include "AnalysisRunner.h"
#include "Globals.h"
#include "SpikeStreamException.h"
using namespace spikestream;

//Qt includes
#include <QMutexLocker>

//Other includes
#include <iostream>
using namespace std;


AnalysisRunner::AnalysisRunner(const DBInfo& netDBInfo, const DBInfo& archDBInfo, const DBInfo& anaDBInfo){
        //Store information about databases
        this->networkDBInfo = netDBInfo;
        this->archiveDBInfo = archDBInfo;
        this->analysisDBInfo = anaDBInfo;

        //Initialize variables
        stopThread = true;
}


AnalysisRunner::~AnalysisRunner(){
}


/*-------------------------------------------------------------*/
/*-------                 PUBLIC METHODS                 ------*/
/*-------------------------------------------------------------*/

void AnalysisRunner::prepareAnalysisTask(const AnalysisInfo& analysisInfo, int firstTimeStep, int lastTimeStep){
        if(subThreadsRunning())
                throw SpikeStreamException("Cannot prepare analysis task when sub threads are running.");
        if(stopThread == false)
                throw SpikeStreamException("Cannot prepare analysis task when analysis thread is running.");

        //Reset class
        this->reset();

        //Check variables
        if(analysisInfo.getID() == 0)
                throw SpikeStreamException("Analysis ID has not been set.");
        if(firstTimeStep > lastTimeStep)
                throw SpikeStreamException("First time step is greater than the last time step.");

        //Store variables
        this->analysisInfo = analysisInfo;
        this->firstTimeStep = firstTimeStep;
        this->lastTimeStep = lastTimeStep;
        nextTimeStep = firstTimeStep;
}


void AnalysisRunner::run(){
        stopThread = false;

        //Launch the first batch of threads
        unsigned int threadCount = 0;
        while(threadCount < analysisInfo.getNumberOfThreads()){
                int tmpTimeStep = getNextTimeStep();

                //No more time steps, break out of the launching loop
                if(tmpTimeStep == -1)
                        break;

                //Start the thread running to analyze this time step
                try{
                        startAnalysisTimeStepThread(tmpTimeStep);
                        ++threadCount;
                }
                catch(SpikeStreamException& ex){
                        setError(ex.getMessage());
                }
        }

        /* Wait for the analysis to complete
        When all of the sub threads have finished, stopThread should be set to true. */
        while(!stopThread){
                usleep(500000);//Sleep for half a second
        }
}


void AnalysisRunner::stop(){
        //Halt all of the other threads and wait for them to finish
        for(QHash<int, AbstractAnalysisTimeStepThread*>::iterator iter = subThreadMap.begin(); iter != subThreadMap.end(); ++iter){
                iter.value()->stopThread();
                iter.value()->wait();
                delete iter.value();
        }

        subThreadMap.clear();
        stopThread = true;
}


void AnalysisRunner::setTimeStepThreadCreationFunction(AbstractAnalysisTimeStepThread* (*createAnalysisTimeStepThread)(const DBInfo& netDBInfo, const DBInfo& archDBInfo, const DBInfo& anaDBInfo)){
        this->createAnalysisTimeStepThread = createAnalysisTimeStepThread;
}


/*-------------------------------------------------------------*/
/*-------                  PRIVATE SLOTS                 ------*/
/*-------------------------------------------------------------*/

void AnalysisRunner::updateProgress(const QString& msg, unsigned int timeStep, unsigned int stepsCompleted, unsigned int totalSteps){
        emit progress(msg, timeStep, stepsCompleted, totalSteps);
}


void AnalysisRunner::threadFinished(){
        //Do nothing if thread is trying to stop
        if(stopThread)
                return;

        /* Multiple sub threads could call this method simultaneously and mess things up
        Lock mutex to make sure we only process one thread finished event at a time. */
        QMutexLocker(&this->mutex);

        //Get a reference to the thread that has stopped
        AbstractAnalysisTimeStepThread* tmpSubThread = (AbstractAnalysisTimeStepThread*) sender();

        //Remove it from the map of currently running threads
        int tmpTimeStep = tmpSubThread->getTimeStep();
        if(!subThreadMap.contains(tmpTimeStep)){
                setError("Thread with time step " + QString::number(tmpTimeStep) + " is not recorded as running.");
                return;
        }

        //Check for errors - set error will delete all of the threads
        if(tmpSubThread->isError()){
                setError(tmpSubThread->getErrorMessage());
                return;
        }

        //Delete the thread class and remove it from the map
        delete tmpSubThread;
        subThreadMap.remove(tmpTimeStep);

        //Inform listening classes that the analysis of this time step is now complete.
        emit timeStepComplete(tmpTimeStep);

        //Determine if any time steps need to be analyzed
        int tmpNextTimeStep = getNextTimeStep();
        if(tmpNextTimeStep >= 0){
                try{
                        startAnalysisTimeStepThread(tmpNextTimeStep);
                }
                catch(SpikeStreamException& ex){
                        setError(ex.getMessage());
                }
        }

        //Next time step is -1. All of the time step threads have been launched and may have all finished
        else if(!subThreadsRunning()){//See if any threads are still running
                //Stop this thread when no sub threads are running
                stop();
        }
}


void AnalysisRunner::updateResults(){
        emit newResultsFound();
}


/*-------------------------------------------------------------*/
/*-------                PRIVATE METHODS                 ------*/
/*-------------------------------------------------------------*/

void AnalysisRunner::clearError(){
        error = false;
        errorMessage = "";
}


int AnalysisRunner::getNextTimeStep(){
        if(nextTimeStep == -1)
                return -1;

        //Store current value of next time step. This is what will be returned
        int oldNextTimeStep = nextTimeStep;

        //Increase the next time step and check to see if it has exceeded the range
        ++nextTimeStep;
        if(nextTimeStep > lastTimeStep)
                nextTimeStep = -1;//No more time steps left. -1 will be returned next time this method is called

        //Return the original value of the next time step before it was increased or invalidated
        return oldNextTimeStep;
}


void AnalysisRunner::reset(){
        subThreadMap.clear();
        stopThread = true;
        clearError();
        firstTimeStep = -1;
        lastTimeStep = -1;
        nextTimeStep = -1;
}


void AnalysisRunner::setError(const QString& message){
        error = true;
        errorMessage = message;

        //exit this thread
        stop();
}


void AnalysisRunner::startAnalysisTimeStepThread(int timeStep){
        //Run some checks on the time step
        if(timeStep == -1){
                throw SpikeStreamException("Trying to start a new thread with an invalid time step.");
        }
        if(subThreadMap.contains(timeStep)){
                throw SpikeStreamException("Trying to launch new thread to analyze time step " + QString::number(timeStep) + " which is already running.");
        }

        //Lanch the new thread and store its address in the map
        AbstractAnalysisTimeStepThread* newThread = createAnalysisTimeStepThread(networkDBInfo, archiveDBInfo, analysisDBInfo);
        connect(newThread, SIGNAL(newResultsFound()), this, SLOT(updateResults()));
        connect(newThread, SIGNAL(finished()), this, SLOT(threadFinished()));
        connect(newThread, SIGNAL(progress(const QString&, unsigned int, unsigned int, unsigned int)), this, SLOT(updateProgress(const QString&, unsigned int, unsigned int, unsigned int)));
        newThread->prepareTimeStepAnalysis(analysisInfo, timeStep);
        subThreadMap[timeStep] = newThread;
        newThread->start();
}


bool AnalysisRunner::subThreadsRunning(){
        if(subThreadMap.isEmpty())
                return false;
        return true;
}