mthread.cpp

// #################################################################################################
//  ALib C++ Library
//
//  Copyright 2013-2019 A-Worx GmbH, Germany
//  Published under 'Boost Software License' (a free software license, see LICENSE.txt)
// #################################################################################################
// todo: add precomp everywhere
 
#if !defined(HPP_AWORX_ALIB_THREADS_MODEL_MTHREAD)
#   include "alib/threads/model/mthread.hpp"
#endif
#if ALIB_ENUMS
#   if !defined(HPP_ALIB_ENUMS_SERIALIZATION)
#      include "alib/enums/serialization.hpp"
#   endif
#endif
 
#if !defined(ALIB_DOX)  // todo remove
 
 
namespace alib::threads::model {
 
MThread::MThread(const String& pName)
: Thread( pName )
, stoppedFlag(false)
{
}
 
MThread::~MThread()
{
    State curState= GetState();
         if( curState == Thread::State::Unstarted )
    {
        ALIB_MESSAGE("ATHR",  NString256( "Thread \"" ) << GetName()
                              << "\" is destructed, while it was never started"  )
    }
    else if( curState != Thread::State::Terminated )
    {
        ALIB_MESSAGE("ATHR",  NString256( "Thread \"" ) << GetName()
                              << "\" is destructed, while it is in state \""
                              << curState << '"' )
        return;
    }
}
 
void MThread::Stop()
{
    stoppedFlag= true;
    cmdPipe.Push(MThread::InternalCommands::Stop, 10, nullptr, nullptr );
}
 
void MThread::Run()
{
    ALIB_MESSAGE("ATHR", NString256( "MThread \"") << GetName() << "\" is running" )
 
    while(!stoppedFlag)
    {
        CmdPipe::Command cmd= cmdPipe.Pop();
 
        // internal commands
        if( cmd.cmd.IsEnumType<InternalCommands>() )
        {
            #if !ALIB_DEBUG
                ALIB_WARNINGS_ALLOW_SPARSE_ENUM_SWITCH
            #endif
            switch (cmd.cmd.Get<InternalCommands>())
            {
                case InternalCommands::Stop:
                    #if ALIB_DEBUG
                        ALIB_MESSAGE("ATHR", NString256("MThread \"") << GetName()
                                              << "\" received stop command" )
                        if( stoppedFlag == false )
                            ALIB_ERROR("ATHR", NString256("MThread \"") << GetName()
                                       << "\": Stop command in pipe, but stop flag not set")
                    #endif
                    break;
 
 
                #if ALIB_DEBUG
                case InternalCommands::NoCommand:
                    ALIB_ERROR("ATHR", "Illegal command 'NoCommand' received.")
                    break;
                #endif
            }
            #if !ALIB_DEBUG
                ALIB_WARNINGS_RESTORE
            #endif
 
            // that's it in respect to internal commands
            continue;
        }
 
        // process external command
        processCommand(cmd);
    }
 
    #if ALIB_DEBUG
        NString256 msg( "MThread \""); msg << GetName() << "\" is stopping.";
        if( cmdPipe.PeekSize() > 0 )
        {
            msg << ' ' << cmdPipe.PeekSize() << "commands still in pipe!" << NewLine();
            for( int i= 0; i < std::min(10, cmdPipe.PeekSize()) ; ++i)
                msg << "   " << i << ": " << cmdPipe.PeekCmd(i).cmd << NewLine();
        }
        ALIB_MESSAGE( "ATHR", msg )
    #endif
}
 
//--------------------------------------------------------------------------------------------------
// MThread::CmdPipe
//--------------------------------------------------------------------------------------------------
void MThread::CmdPipe::Push(Enum cmd, int priority, Box data, Box data2)
{
    { ALIB_LOCK
        // search the first element with equal or higher priority
        std::list<Command>::iterator   it=   pipe.begin();
        while(it != pipe.end())
        {
            if(priority <= (*it).priority)
                break;
            it++;
        }
 
        // insert before found
        it= pipe.insert(it, Command(cmd,priority,data, data2));
 
        // statistics
        usageCounter++;
    } // lock
 
    // if consumer is sleeping then wake him up
    Notify();
}
 
MThread::CmdPipe::Command  MThread::CmdPipe::Pop()
{
    // threadsafe access to the pipe
    Acquire(ALIB_CALLER_PRUNED);
 
        // loop until we get a command (the Acquire after sleep is
        // unsafe, therefore we have to loop as another thread may
        // steal the message that waked us up meanwhile....
        while(pipe.size()==0)
        {
            // no, -> sleep!
            WaitForNotification();
            Acquire(ALIB_CALLER_PRUNED);
            #if ALIB_DEBUG
                if( pipe.size() == 0)
                    ALIB_WARNING("ATHR", "Cmd pipe empty after wakeup" )
                if( pipe.size() > 2)
                    ALIB_WARNING("ATHR", NString256("Cmd pipe growing to: ") << pipe.size() )
            #endif
        }
 
        // get message
        Command msg(pipe.back());
        pipe.pop_back();
 
    Release();
 
    // log thread activity
    return msg;
}
 
MThread::CmdPipe::Command     MThread::CmdPipe::PeekCmd(int n)
{
    auto it= pipe.begin();
    for( int i= 0; i < n ; ++i)
        if( it!=pipe.end() )
            ++it;
    return it != pipe.end() ? *it : Command( InternalCommands::NoCommand, 0, nullptr, nullptr );
}
 
 
 
} // namespace [alib::threads::model]
 
#endif // !defined(ALIB_DOX)