src/MessageController.cpp

Go to the documentation of this file.

/*

Traplas visualisation.

Copyright (C) 2006 Herbert de Vos & Willem Drost

This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

(For full Licence see ../GPL-Licence.txt)

*/

#ifndef _GLIBCXX_VECTOR
#include <vector>
#endif

#ifndef _GLIBCXX_STRING
#include <string>
#endif

#ifndef _GLIBCXX_IOSTREAM
#include <iostream>
#endif

#ifndef _GLIBCXX_SSTREAM
#include <sstream>
#endif

#ifndef _TIME_H
#include <time.h>
#endif

#ifndef _SYS_TIMEB_H
#include <sys/timeb.h>
#endif

#ifndef MESSAGECONTROLLER_H
#include "MessageController.h"
#endif

#ifndef DISPATCHER_H
#include "Dispatcher.h"
#endif

#ifndef FILEREADER_H
#include "FileReader.h"
#endif

#ifndef DBUSCONNECTION_H
#include "DbusConnection.h"
#endif


#ifndef GLOBALS_H
#include "Globals.h"
#endif

#ifndef _GLIBCXX_STDEXCEPT
#include <stdexcept>
#endif

long time_millisec()
{
        long tm = 0;
        struct timeb *t = new struct timeb;
        if ( ftime(t) == -1 )
                        cout << "Error getting time." << endl;
        tm = 0;
        tm += ((long)t->time) * (long)1000;
        tm += t->millitm;
        delete t;
        return tm;
}

void* msgcontrollerThread(void *args)
{
        MessageController *myc = (MessageController*) args;

        double lasttime = time_millisec();
        
        bool done;
        double nextmessagetime = 0;
        double newtime;
        string temp;
        double messagetimestamp;

        while( myc->msgcontrollerthread_run ) {
                done = false;
                
                //      Update time according to the timeflowfactor.
                newtime = time_millisec();
                myc->currenttime += ((newtime - lasttime) * myc->timeflowfactor);
                lasttime = newtime;
                
                if ( pthread_mutex_trylock( &(myc->mutex) ) != 0 ) 
                        pthread_cond_wait( &(myc->cond) , &(myc->mutex) );
                                
                //      Check if there are messages to be sent.
                if( !myc->msgqueue->empty() && (nextmessagetime <= myc->currenttime/1000) ) {
                        //      Decide if message from the back of the queue must be send. (back of the queue have the lowest timestamp!)
                        temp = *(myc->msgqueue->back()->at(1));
                        messagetimestamp = atof( temp.c_str() );
                        
                        #if STRICT
                        //Check if conversion has worked.
                        assert( !(messagetimestamp == 0 && temp.at(0) != '0') );
                        #endif
                        
                        if ( messagetimestamp <= myc->currenttime/1000 ) {
                                
                                myc->disp->proccessMessageFromMC( myc->msgqueue->back() );
                                myc->msgqueue->pop_back();

                        } else {
                                nextmessagetime = messagetimestamp;
                                done = true;
                        }
                } else
                        done = true;

                //      Release mutex, signal.
                pthread_mutex_unlock( &(myc->mutex) );
                pthread_cond_signal( &(myc->cond) );
                //      And sleep.
                if ( done )
                        usleep(50);     // Nothing more to be done at this time, go sleep.
        }

        cout << "Messagecontroller thread exit." << endl;
        int status = 0;
        pthread_exit(&status);
}

MessageController::MessageController(Dispatcher* dp) {
        comm = NULL;
        disp = dp;
        msgqueue = shared_ptr< list< shared_ptr< vector< shared_ptr<string> > > > >( new list< shared_ptr< vector< shared_ptr<string> > > >() );

        pthread_mutex_init(&mutex, NULL);       
        pthread_cond_init(&cond, NULL);

        currenttime = 0;
        timeflowfactor = 0;

        msgcontrollerthread_run = true;
        pthread_create(&msgcontrollerthread, NULL, msgcontrollerThread, (void*) this);
}

MessageController::~MessageController() {
        msgcontrollerthread_run = false;        //      Signal the msgcontrollerthread to stop.

        if ( comm != NULL )
                delete comm;

        pthread_join( msgcontrollerthread, NULL);
}

bool MessageController::processMessageFromComm(shared_ptr< vector< shared_ptr< string > > > messagebody) {
        //      Adds the message to the queue at the correct place so the queue remains ordered by timestamp. The first element of the queue is the element with the highest timestamp and the last is with the lowest timestamp. This is because it is impossible to insert while using a reverse_iterator.

        string type = *(messagebody->at(0));
        string ts = *(messagebody->at(1));
        
        double msg_timestamp = atof( ts.c_str() );
        
        #if STRICT
        //      Check if conversion has worked.
        assert( !(msg_timestamp == 0 && ts.at(0) != '0') );     
        #endif
        
        //      Check the type of the message and take appropriate action.
        if ( type == "BEGIN" ) {                                                //      BEGIN message
                timeflowfactor = INITIAL_TIMESCALE;
        } else {                                                                                //      Put message in the queue.
                if ( pthread_mutex_trylock( &mutex ) != 0 )
                        pthread_cond_wait( &cond , &mutex );

                if ( !msgqueue->empty() ) {
                        bool done = false;
                        list< shared_ptr< vector< shared_ptr< string> > > >::iterator iter;
                        iter = msgqueue->begin();
                        while( !(iter == msgqueue->end() || done) )
                        {
                                double queue_msg_timestamp = atof ( (*((*iter)->at(1))).c_str() );
        
                                if ( msg_timestamp >= queue_msg_timestamp ) {
                                        msgqueue->insert(iter, messagebody);
                                        done = true;
                                }
                                iter++;
                        }
                        if (!done) {
                                //      Message timestamp is the lowest thusfar.
                                msgqueue->push_back(messagebody);
                        }
                } else {
                        //      Queue is empty.
                        msgqueue->push_front(messagebody);
                }
        
                //      Release mutex and signal.
                pthread_mutex_unlock(&mutex);
                pthread_cond_signal(&cond);
        }

        return true;
}

bool MessageController::sendMessageToComm(shared_ptr< vector< shared_ptr< string > > > messagebody) {
        //      Verify that a communication-object exists.
        assert (comm != NULL);
        
        //      Add timestamp to the message.
        stringstream s;
        s << currenttime / 1000;

        shared_ptr< string > timestamp( new string( s.str() ) );

        try{
                messagebody->at(1) = timestamp;
        } catch ( out_of_range ) {
                cout << "Warning: (MessageController::sendMessageToComm) message malformed: Message too short. Aborting send." << endl;
                return false;
        }

        //      Send the message.
        return comm->sendMessage(messagebody);
}

bool MessageController::communicationMode( int cMode , string filename) {
        // We are going to read from file.
        comm = new FileReader( filename , this );;
        cmode = cMode;

        return true;
}

bool MessageController::communicationMode( int cMode , string thisname , string thatname ) {
        // Using Dbus.
        comm = new DbusConnection( thisname , thatname , this );
        cmode = cMode;

        return true;
}

bool MessageController::setTimeFlow(double ntimeflowfactor) {
        timeflowfactor = ntimeflowfactor;
        
        return true;
}

---