threadmanager.cpp

// #################################################################################################
//  ALib C++ Library
//
//  Copyright 2013-2019 A-Worx GmbH, Germany
//  Published under 'Boost Software License' (a free software license, see LICENSE.txt)
// #################################################################################################
#if !defined(HPP_AWORX_ALIB_THREADS_MODEL_THREADMANAGER)
#   include "alib/threads/model/threadmanager.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 {
 
ThreadManager::ThreadManager()
: Thread("ThrdMgr")
{
    mStoppedFlag= false;
}
 
ThreadManager::~ThreadManager()
{
    if( state <= State::Started )
    {
        ALIB_WARNING("ATHR", "ThreadManager destroyed without having run" )
    }
    else if( state != State::Terminated )
    {
        ALIB_ERROR("ATHR", "ThreadManager destroyed without being terminated" )
        Start();
    }
}
 
void ThreadManager::Add(MThread& thread)
{
    std::unique_lock<std::mutex> lock(mutex);
    threadMap.emplace( thread.GetId() , &thread );
    thread.Start();
}
 
 
void ThreadManager::StopAndTerminateAll()
{
    ALIB_MESSAGE("ATHR", "ThreadManager::StopAndTerminateAll" )
 
    // stop myself
    if( state < State::Stopped)
    {
        mStoppedFlag= true;
        NotifyAll();
    }
 
    ALIB_DBG( Ticks waitCheck= Ticks::Now();
              int nextWarnSecond= 1; )
    while( state > State::Unstarted && state < State::Stopped )
    {
        Thread::SleepMicros( 50 );
        ALIB_DBG(   if( waitCheck.Age().InAbsoluteSeconds() == nextWarnSecond )
                        ALIB_WARNING( "ATHR", "Waiting for ThreadManager to stop")           )
    }
 
    // stop all registered MThreads
    { ALIB_LOCK
 
        // send stop to those unstopped
        for( auto it : threadMap )
            if( !it.second->IsStopped() )
                it.second->Stop();
 
        ALIB_DBG( waitCheck= Ticks::Now(); nextWarnSecond= 1; )
        int cntRunning;
        while(true)
        {
            // check that all threads are stopped
            cntRunning= 0;
            for( auto it : threadMap )
                if( it.second->GetState() < Thread::State::Stopped  )
                    cntRunning++;
            if( cntRunning == 0 )
                break;
 
            Thread::SleepMicros( 50 );
            #if ALIB_DEBUG
                if( waitCheck.Age().InAbsoluteSeconds() == nextWarnSecond )
                {
                    NString4K dbgThreadList("ThreadManager Termination: Waiting on " );
                    dbgThreadList << static_cast<integer>(cntRunning) << " Threads to stop. Threadlist: " << NewLine();
                    int tNr= 0;
                    for( auto it : threadMap  )
                    {
                        dbgThreadList << ++tNr << ": " << it.second->GetName() << ",\tState::"
                        #if ALIB_ENUMS
                                               <<                    it.second->state  << NewLine();
                        #else
                                               <<  static_cast<int>( it.second->state) << NewLine();
                        #endif
                    }
                ALIB_WARNING( "ATHR", dbgThreadList.NewLine().Terminate() )
                    nextWarnSecond++;
                }
            #endif
        }
 
        // terminate all registered MThreads
        for( auto it : threadMap )
                it.second->Terminate();
        threadMap.clear();
 
        // terminate myself
        Terminate();
    } // lock
}
 
 
void ThreadManager::Run()
{
    // ---------------------   Trigger loop   ---------------------
    Ticks nextWakeup= Ticks::Now();
    ALIB_DBG( Ticks::Duration dbgSleepTime; )
    while(!IsStopped())
   {
        Ticks now= Ticks::Now();
        
        #if ALIB_DEBUG
            auto diff= (nextWakeup - now).InAbsoluteMicroseconds();
            NString256 msg("ThreadManager woke up. Scheduled wakeup " );
            if( diff < 0 )
            {
                msg << "was " << -diff << "\u00B5s ago ("
                    << static_cast<double>( -100 * diff / dbgSleepTime.InAbsoluteMicroseconds())
                       / 100.0
                    << "% late)";
            }
            else
            {
                msg << "is in " << diff << "\u00B5s ("
                    << static_cast<double>( 100 * diff  / dbgSleepTime.InAbsoluteMicroseconds())
                       / 100.0
                    << "% early)";
            }
            ALIB_MESSAGE("ATHR", msg.Terminate())
        #endif
 
            //--------- trigger triggered objects --------
            { ALIB_LOCK
                for( auto it : triggerList )
                {
                    if( it.nextWakeup <= now )
                    {
                        it.triggered->trigger();
                        now= Ticks::Now();
                        it.nextWakeup= now + Ticks::Duration::FromAbsoluteMilliseconds(it.triggered->triggerPeriodInMilliseconds());
                    }
 
                    if( nextWakeup < it.nextWakeup)
                        nextWakeup= it.nextWakeup;
                }
            }
 
            // quickly check for stopped signal
            if( IsStopped() )
            {
                ALIB_MESSAGE("ATHR",  "ThreadManager Stopping")
                break;
            }
 
            // sleep and update status for threadDB
            ALIB_DBG( dbgSleepTime= nextWakeup - now; )
            SleepUntil( nextWakeup );
    }
 
     ALIB_MESSAGE("ATHR", "ThreadManager Exiting" )
}
 
void ThreadManager::AddTriggered(Triggered& triggered)
{
    { ALIB_LOCK
 
        // find if not already registered
        auto it= triggerList.begin();
        while(it != triggerList.end() && it->triggered != &triggered);
 
        if( it == triggerList.end() )
            triggerList.push_back( { &triggered,
                                        Ticks::Now() + Ticks::Duration::FromAbsoluteMilliseconds(
                                                  triggered.triggerPeriodInMilliseconds() )
                                       } );
        else
            it->nextWakeup= Ticks::Now(); // we do a first wakeup as soon as possible away
    }
    // let's wake this thread up and perform a first trigger
    Notify();
}
 
void ThreadManager::RemoveTriggered(Triggered &triggered)
{
    bool found= false;
 
    { ALIB_LOCK
        for(auto it= triggerList.begin() ; it != triggerList.end() ; it++)
        {
            if(it->triggered == &triggered)
            {
                found= true;
                triggerList.erase(it);
                break;
            }
        }
    }
 
    // check
    if(!found)
    {
        ALIB_MESSAGE("ATHR", NString256( "Managed thread \"" )
                             << triggered.Name
                             << "\" not found for de-registering with trigger list" )
    }
}
 
 
 
 
} // namespace [alib::threads::model]
 
#endif // !defined(ALIB_DOX)