source: trunk/MagicSoft/Cosy/base/msgqueue.cc@ 8611

#include "msgqueue.h"

#include <iostream>

#include <unistd.h>        // usleep
#include <sys/resource.h>  // PRIO_PROCESS

#undef DEBUG

using namespace std;

// --------------------------------------------------------------------------
//
// This creates the Message queue thread,
//
MsgQueue::MsgQueue() : fBreak(0)
{
    fMp = new unsigned char;
    pthread_create(&fThread, NULL, MapThread, this);
}

// --------------------------------------------------------------------------
//
// The destructor terminates the thread.
//
MsgQueue::~MsgQueue()
{
#ifdef DEBUG
    cout << "~MsgQueue::MsgQueue" << endl;
#endif
    pthread_cancel(fThread);
    delete (unsigned char*)fMp;
}

// --------------------------------------------------------------------------
//
// This is the function which must be overloaded.
// Here you process the messages. mp is a pointer which you can
// specify when posting the message.
//
// If a new messages is posted while the old one is not yet
// finished the fBreak flag is set. Please test this flag with
// Break() and try to finish (or stop) the current action as soon
// as possible. This makes sure, that before a new action is started
// the old action can be finished correctly by the user.
//
void *MsgQueue::Proc(int msg, void *mp)
{
    return NULL;
}

// --------------------------------------------------------------------------
//
// This is the thread mapper.
//
void *MsgQueue::MapThread(void *arg)
{
    pthread_detach(pthread_self());

    setpriority(PRIO_PROCESS, 0, -5);

    ((MsgQueue*)arg)->Thread();

    return NULL;
}

// --------------------------------------------------------------------------
//
// This is the thread which handles the processing.
// As soon as a message is posted the fBreak flag is set (see PostMsg)
// And as soon as the current action is finished the new action is executed
// in this thread. This makes sure, that the calling program is not stalled.
//
void MsgQueue::Thread()
{
    //
    // Tell the poster that processing is done
    //
    fStart = 0;
    while (!fBreak)
        usleep(1);

    while(1)
    {
        while (!fStart)
            usleep(1);
        fStart = 0;

        //
        // This makes sure that also a very fast Break() after
        // a PostMsg is processed correctly
        //
        if (fMuxMsg.Lock()==13)
            cout << "MsgQueue::Thread - mutex is already locked by this thread." << endl;
        fBreak = 0;
        if (fMuxMsg.UnLock()==13)
            cout << "MsgQueue::Thread - tried to unlock mutex locked by other thread." << endl;

#ifdef DEBUG
        cout << "MsgQueue::Thread: Processing Msg 0x" << hex << fMsg << endl;
#endif
        // --- bool quit = fMsg==WM_QUIT;
        fRc=Proc(fMsg, fMp);
#ifdef DEBUG
        cout << "MsgQueue::PostMsg: Msg 0x" << hex << fMsg << " processed (rc=" << fRc << ")" << endl;
#endif

        // --- if (quit)
        // ---    break;
    }

    // --- fStart = 0;
    // --- cout << "WM_QUIT posted... leaving msg queue." << endl;
}

// --------------------------------------------------------------------------
//
// Use this function to post a message.
// mp can be a pointer to a data structure. size should be the size of it.
// size bytes of this structure are copied and a pointer to the copy
// is forwarded to the Proc function.
//
void *MsgQueue::PostMsg(int msg, void *mp, int size)
{
    //
    // Lock Mutex, put msg on stack and tell thread to process message
    //

    //
    // Make sure that only one Proc() is running and can be stopped
    // stopped and the messages are processed serialized
    //
#ifdef DEBUG
    cout << "MsgQueue::PostMsg: Locking MsgQueue mutex..." << flush;
#endif
    if (fMuxMsg.Lock()==13)
        cout << "MsgQueue::PostMsg - mutex is already locked by this thread." << endl;
#ifdef DEBUG
    cout << "done." << endl;
#endif

    //
    // Set break state and wait until Proc() returned (break state deleted)
    // This means, that a new command is invoked and (if forseen) the
    // running command should stop execution.
    //
    // This is some kind of controlled user break without using signals
    //
    /**** NEW 20/01/2003 ****/
    if (fBreak)
    {
        if (fMuxMsg.UnLock()==13)
            cout << "MsgQueue::PostMsg - tried to unlock mutex locked by other thread." << endl;
#ifdef DEBUG
        cout << "------------> MsgQueue::PostMsg: Proc still pending... Message IGNORED." << endl;
#endif
        return NULL;
    }
    /**** NEW 20/01/2003 ****/
#ifdef DEBUG
    cout << "MsgQueue::PostMsg: ---> Break <---" << endl;
#endif
    fBreak = 1;

    //
    // copy return code from Proc() and set new message
    //
    void *rc = fRc;

    fMsg = msg;

    delete (unsigned char*)fMp;
    fMp = new unsigned char[size];

    memcpy(fMp, mp, size);

    //
    // Start Proc()
    //
#ifdef DEBUG
    cout << "MsgQueue::PostMsg: Releasing MsgQueue mutex..." << flush;
#endif
    fStart = 1;
    if (fMuxMsg.UnLock()==13)
        cout << "MsgQueue::PostMsg - tried to unlock mutex locked by other thread." << endl;
#ifdef DEBUG
    cout << "done." << endl;
#endif

    /*
     * **** NEW 20/01/2003 ***
     *
     * This can halt the main thread, because it is waiting until
     * Proc has finished its execution which can take a while
     *
     * A side effect is, because you don't wait for the end of
     * the execution of Proc, if a new message is posted fBreak
     * and fStart is set again, new values are copied to fMsg and
     * fMp (FIXME? Harmefull?) and the message is not processed at all.
     */
    //while (fStart)
    //    usleep(1);
 
#ifdef DEBUG
    cout << "MsgQueue::PostMsg: Returning rc = " << hex << rc << endl;
#endif
    return rc;
}
/*
Start positioning.
MsgQueue::PostMsg: Locking MsgQueue mutex...done.
MsgQueue::PostMsg: ---> Break <---
+++++ MsgQueue::PostMsg: Releasing MsgQueue mutex...done.

===== MsgQueue::PostMsg: Returning rc = (nil)
PostMsg (WM_Position) returned.
done.
Stopping movement...Movement stopped.
WM_Position: done. (return 0x7777)
MsgQueue::PostMsg: Msg 0x1001 processed (rc=0x7777)
MsgQueue::Thread: Processing Msg 0x1001
WM_Position: start.
Positioning to Target...
 */
/*
Start positioning.
MsgQueue::PostMsg: Locking MsgQueue mutex...done.
MsgQueue::PostMsg: ---> Break <---
+++++ MsgQueue::PostMsg: Releasing MsgQueue mutex...done.

done.
Stopping movement...Movement stopped.
WM_Position: done. (return 0x7777)
MsgQueue::PostMsg: Msg 0x1001 processed (rc=0x7777)
MsgQueue::Thread: Processing Msg 0x1001
WM_Position: start.
Positioning to Target...
===== MsgQueue::PostMsg: Returning rc = (nil)
*/