Thread.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/Utilities/Thread.h>

#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/ThreadPriorityConfiguration.h>
#include <fesa-core/Exception/FesaException.h>
#include <fesa-core/Utilities/Lock.h>
#include <fesa-core/Utilities/Mutex.h>
#include <fesa-core/Utilities/ProcessConfiguration.h>
#include <fesa-core/Utilities/XMLParser.h>

#include <cmw-log/Logger.h>

#include <cstring>
#include <sys/resource.h>
#include <sys/syscall.h>
#include <sys/types.h>


namespace
{

CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.Utilities.Thread");

int32_t process_priority = 0;
int32_t scheduling_policy = fesa::RT_SCHED_POLICY;
bool runWithoutRTPrios = false;
std::map<pthread_t, std::string> threadIdNameMap;
fesa::Mutex threadIdNameMapMutex;


int
getProcessPriority()
{
    // To get the priority of a process, you have to get the priority from the mainthread of this process.
    // So this method only is called from the MainThread.
    pid_t pid = getpid();
    sched_param params;
    if (sched_getparam(pid, &params) == -1)
    {
        std::string errMsg(std::strerror(errno));
        std::ostringstream errorStrStream;
        errorStrStream << "Could not get Process Parameter: " << errMsg;
        throw fesa::FesaException(__FILE__, __LINE__, FesaErrorReadProcessParameter.c_str(), errMsg.c_str());
    }
    process_priority = params.sched_priority;
    return params.sched_priority;
}


void
setProcessPriority(int32_t prio, int32_t policy)
{
    //To adjust the priority of a process, you have to adjust the priority from the mainthread of this process.
    //So this Method only is called from the MainThread.
    pid_t pid = getpid();
    sched_param params;
    sched_getparam(pid, &params);
    params.sched_priority = prio;
    int32_t err = sched_setscheduler(pid, policy, &params);
    if (err != 0)
    {
        std::string errMsg(std::strerror(err));
        throw fesa::FesaException(__FILE__, __LINE__, FesaErrorWriteProcessParameter.c_str(), errMsg.c_str());
    }
    std::ostringstream stream;
    stream << "Process priority set to priority: " << prio;
    stream << ", Scheduling policy set to: ";
    if(policy == fesa::RT_SCHED_POLICY)
    {
        stream << "SCHED_RR" << std::endl;
    }
    else
    {
        stream << "SCHED_OTHER" << std::endl;
    }
    LOG_TRACE_IF(logger, stream.str());
    process_priority = prio;
}

} // namespace


namespace fesa
{

void*
Thread::startThread(void* arg)
{
    Thread* thread = reinterpret_cast<Thread*>(arg);
    if(runWithoutRTPrios)
    { // Nice level needs to be set at run-time
        // That's how a nice-level can be set.
        // We can't supply thread->tid_ as the second parameter. 'setpriority' wants the thread id in the form that 'gettid()' provides,
        // which is different from the one returned from 'pthread_create()'
        if (setpriority(PRIO_PROCESS, static_cast<pid_t>(syscall(SYS_gettid)), thread->priority_) == -1)
        {
            const std::string errMsg(std::strerror(errno));
            throw FesaException(__FILE__, __LINE__, FesaErrorSettingThreadAttributes.c_str(),errMsg.c_str());
        }
    }
    registerThreadIdName(thread->tid_, thread->name_);
    thread->run();

    return 0;
}


Thread::Thread() :
    isRunning_(false),
    hasFinished_(false)
{
    priority_ = -100;//has to be adjusted by the derived class.

    if(runWithoutRTPrios)
        scheduling_policy = NICE_SCHED_POLICY;
    if (pthread_attr_init(&attr_) != 0)
        throw FesaException(__FILE__, __LINE__, FesaErrorInitializingThreadAttributes.c_str());
}


Thread::~Thread()
{
}


void
Thread::start(bool blockingMode, const std::string& threadName)
{
    if (priority_ == -100) //priority has to be adjusted before startup
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorUndefinedPriority.c_str());
    }
    name_ = threadName;
    const pid_t pid = getpid();
    sched_param params;
    sched_getparam(pid, &params);
    std::ostringstream traceStrStream;
    traceStrStream << "Starting new thread with priority " << priority_;
    if (runWithoutRTPrios == true)
    {
        traceStrStream << ". Attention! Thread is started without RT scheduling" << std::endl;
    }
    LOG_TRACE_IF(logger, traceStrStream.str());
    if (runWithoutRTPrios == true) // Use nice levels instead of priorities
    {
        params.sched_priority = 0;
    }
    else
    {
        params.sched_priority = priority_;
    }
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
    int32_t err = pthread_attr_setschedpolicy(&attr_, scheduling_policy);
    if (err != 0)
    {
        const std::string errMsg(std::strerror(err));
        throw FesaException(__FILE__, __LINE__, FesaErrorSettingSchedulingPolicy.c_str(), errMsg.c_str());
    }
    err = pthread_attr_setinheritsched(&attr_, PTHREAD_EXPLICIT_SCHED);
    if (err != 0)
    {
        const std::string errMsg(std::strerror(err));
        throw FesaException(__FILE__, __LINE__, FesaErrorSettingInheritScheduling.c_str(), errMsg.c_str());
    }
    err = pthread_attr_setschedparam(&attr_, &params);
    if (err != 0)
    {
        const std::string errMsg(std::strerror(err));
        throw FesaException(__FILE__, __LINE__, FesaErrorSettingThreadAttributes.c_str(), errMsg.c_str());
    }
    if(blockingMode)
    {
        pthread_attr_setdetachstate(&attr_, PTHREAD_CREATE_JOINABLE);
    }
    else
    {
        pthread_attr_setdetachstate(&attr_, PTHREAD_CREATE_DETACHED);
    }
    //there is some kind of race with isRunning_
    isRunning_ = true;
    err = pthread_create(&tid_, &attr_, Thread::startThread, reinterpret_cast<void*>(this));
    if (err != 0)
    {
        const std::string errMsg(std::strerror(err));
        throw FesaException(__FILE__, __LINE__, FesaErrorCreatingThread.c_str(),errMsg.c_str());
    }
    isRunning_ = true;
    hasFinished_ = false;
    err = pthread_attr_destroy(&attr_);
    if (err != 0)
    {
        const std::string errMsg(std::strerror(err));
        throw FesaException(__FILE__, __LINE__, FesaErrorDestroyingThreadAttributes.c_str(), errMsg.c_str());
    }
    // If it is a blocking call the calling thread waits for the return of the new thread
    if (blockingMode)
    {
        pthread_join (tid_, NULL);
    }
}


void
Thread::stop()
{
    isRunning_ = false;
    int32_t attempts = MAX_ATTEMPTS_STOP_THREAD;

    //wait for the thread to stop
    while (!hasFinished_ && attempts > 0)
    {
        attempts--;
        sleep(1);
    }
    if (attempts == 0)
        pthread_cancel(tid_);
}


int32_t
Thread::join(void** status)
{
    return pthread_join(tid_, status);
}


void
Thread::setPriority(const int32_t priority)
{
    if(isRunning_)
        throw FesaException(__FILE__, __LINE__, FesaErrorSettingPrioDuringRuntime.c_str());

    priority_  = priority;
}


/*static*/
void
Thread::AdjustProcessPriority()
{
    AbstractEquipment* eqp = AbstractEquipment::getInstance();
    runWithoutRTPrios = eqp->getProcessConfiguration()->isDefined(PropertyTag::NO_RT_SCHEDULING);
    if(runWithoutRTPrios)
    {
        std::ostringstream errorStrStream;
        errorStrStream << std::endl;
        errorStrStream << "------------------------------------------ WARNING !!!! ------------------------------------------"  << std::endl;
        errorStrStream << "You are running this FESA-binary without any rt-support, using nice-priorities "                     << std::endl;
        errorStrStream << "The binary will continue to run in verbose mode, but without any guarantee."                         << std::endl;
        errorStrStream << "To be able to start the class with proper RT-priorities, please contact your system administrator."  << std::endl;
        errorStrStream << "--------------------------------------------------------------------------------------------------"  << std::endl;
        LOG_WARNING_IF(logger, errorStrStream.str());
    }

    std::string instanceFile =eqp->getDeviceDataFileName();
    XMLParser xmlParser(instanceFile, false);
    ThreadPriorityConfigurationFromFile conf(xmlParser, eqp->getProcessConfiguration());
    int32_t desired_process_prio = -100;//just something very small
    int32_t current_process_prio = getProcessPriority();

    if(!conf.getMaximumPrio(desired_process_prio))
    {
        // If there is no prio defined in the instantiation-file, we just keep our current process prio
        desired_process_prio = current_process_prio;
        //TODO: What should be the prio of the main.-thread?
    }

    // Try to adjust the process priority, if necessary
    if(runWithoutRTPrios)
    {//nice priorities have the range [-20 20] maximum is -20
        //The scheduler-prio always has to be 0, when using nice-priorities
        setProcessPriority(0,NICE_SCHED_POLICY);
    }
    else
    {//rt-priorities have the range [0 99] maximum is 99
        if (current_process_prio < desired_process_prio)
        {
            setProcessPriority(desired_process_prio,RT_SCHED_POLICY);
        }
    }
}


/*static*/
std::string
Thread::getThreadName(pthread_t threadId)
{
    Lock lock(threadIdNameMapMutex);
    std::map<pthread_t, std::string>::const_iterator iter = threadIdNameMap.find(threadId);
    if (iter == threadIdNameMap.end())
    {
        std::ostringstream errorStrStream;
        errorStrStream << "Thread with id " << static_cast<uint32_t>(threadId) << " is not registered";
        throw FesaException(__FILE__, __LINE__, errorStrStream.str());
    }
    return iter->second;
}


/*static*/
void
Thread::registerThreadIdName(pthread_t threadId, const std::string& threadName)
{
    Lock lock(threadIdNameMapMutex);
    threadIdNameMap.insert(std::pair<pthread_t, std::string>(threadId, threadName));
}

} // fesa