PeriodicTask.h

Go to the documentation of this file.

/*
* This file is part of ArmarX.
*
* Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
*
* ArmarX is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation.
*
* ArmarX 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, see <http://www.gnu.org/licenses/>.
*
* @package    ArmarXCore::core
* @author     Kai Welke (welke at kit dot edu)
* @date       2012
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
 
#pragma once
 
 
#include <time.h>
 
#include <functional>
#include <mutex>
 
#include <IceUtil/Time.h>
#include <IceUtil/Timer.h>
 
#include <ArmarXCore/core/application/Application.h>
#include <ArmarXCore/core/exceptions/Exception.h>
#include <ArmarXCore/core/time/TimeUtil.h>
#include <ArmarXCore/core/time/Timer.h>
 
#include "ThreadList.h"
#include <cxxabi.h>
 
namespace armarx
{
    /**
     * \class PeriodicTask
     * \ingroup Threads
     * The periodic task executes one thread method repeatedly using the time period specified in the constructor.
     */
    template <class T>
    class ARMARXCORE_IMPORT_EXPORT PeriodicTask :
        virtual public IceUtil::TimerTask,
        virtual public Logging,
        virtual protected PeriodicTaskIceBase
    {
    public:
        /**
         * Typedef for the periodic method. Thread methods need to follow the template
         * void methodName(void). Parameter passing is not used since methods are members
         * of the class.
         */
        typedef void (T::*method_type)(void);
 
        /**
         * Shared pointer type for convenience.
         */
        using pointer_type = IceUtil::Handle<PeriodicTask<T>>;
 
        /**
         * Constructs a periodic task within the class parent which calls the peridoicFn in
         * a new thread.
         *
         * @param parent the class wheter periodicFn is member
         * @param peridicFn reference to the periodic function implementation
         * @param periodMs period time in milliseconds
         * @param assureMeanInterval If set to TRUE, the scheduler tries to
         * compensate timeslot exceedings overtime and not only on the next interval
         * @param name of this thread, that describes what it is doing
         * @param forceSystemTime If set to false, the Timeserver will be used for timing (if available)
         *
         */
        PeriodicTask(T* parent,
                     method_type periodicFn,
                     int periodMs,
                     bool assureMeanInterval = false,
                     std::string name = "",
                     bool forceSystemTime = true)
        {
            shutdown = false;
            running = false;
            functionExecuting = false;
            this->assureMeanInterval = assureMeanInterval;
            executionCounter = 0;
            last_cpu_total_time = 0;
            this->periodicFn = periodicFn;
            this->parent = parent;
 
            if (name.empty())
            {
                char* demangled = nullptr;
                int status = -1;
                demangled = abi::__cxa_demangle(typeid(T).name(), nullptr, nullptr, &status);
                setName(demangled);
                free(demangled);
            }
            else
            {
                setName(name);
            }
 
            intervalMs = periodMs;
            interval = IceUtil::Time::milliSeconds(periodMs);
            timer = new armarx::Timer(forceSystemTime);
            delayWarningTolerance = IceUtil::Time::milliSeconds(5);
 
            workloadList.resize(100, 0.f);
            setTag("PeriodicTask" + this->name);
 
            if (customThreadList)
            {
                customThreadList->addPeriodicTask(this);
            }
 
            appThreadList = ThreadList::getApplicationThreadList();
 
            if (appThreadList)
            {
                appThreadList->addPeriodicTask(this);
            }
        }
 
        /**
         * Destructor stops the thread and waits for completion.
         */
        ~PeriodicTask() override
        {
            stop();
            if (appThreadList)
            {
                appThreadList->removePeriodicTask(this);
            }
 
            if (customThreadList)
            {
                customThreadList->removePeriodicTask(this);
            }
        }
 
        /**
         * @brief Set the name identifying PeriodicTask instances.
         * @param name string containing the name
         */
        void
        setName(std::string name)
        {
            this->name = name;
        }
 
        /**
         * @brief This method sets \p threadList as PeriodicTask::customThreadList and appends \p this to the list.
         * @param threadList a pointer to the ThreadList instance.
         */
        void
        setThreadList(ThreadListPtr threadList)
        {
            this->customThreadList = threadList;
 
            if (isRunning())
            {
                threadList->addPeriodicTask(this);
            }
        }
 
        /**
         * Starts the thread. Only has effect if the task has not been started. After completion of the task, it can be started again (see isRunning()).
         */
        void
        start()
        {
            std::unique_lock lock(dataMutex);
            auto app = Application::getInstance();
            if (app && app->getForbidThreadCreation())
            {
                throw LocalException()
                    << "Thread creation is now allowed in this application at the point in time! "
                       "Use Application::getInstance()->getThreadPool() instead.";
            }
            if (!running)
            {
                shutdown = false;
                cycleStart = logTime = __now();
                startTime = logTime.toMicroSeconds();
                timeExceeded = false;
                timer->schedule(this, IceUtil::Time::milliSeconds(0));
            }
 
            running = true;
        }
 
        /**
         * @return The scheduling interval of the thread instance in milliseconds
         */
        IceUtil::Int64
        getInterval() const
        {
            return interval.toMilliSeconds();
        }
 
        /**
         * @return Set the tolerance value for delay warnings. If the tasks takes longer than (scheduling interval + tolerance), a warning is issued.
         * @param newToleranceInMilliSeconds The tolerance, in milliseconds.
         */
        void
        setDelayWarningTolerance(int newToleranceInMilliSeconds = 5)
        {
            delayWarningTolerance = IceUtil::Time::milliSeconds(newToleranceInMilliSeconds);
        }
 
        /**
         * @brief changeInterval changes the interval at which the task is executed.
         * The execution starts the first time again after the duration specified
         * in the parameter OR if the task is currently executed after this task
         * + the interval specified - cycleDuration.
         * @param intervalInMs new execution interval in milliseconds.
         */
        void
        changeInterval(unsigned int intervalInMs)
        {
            std::unique_lock lock(dataMutex);
 
            if (interval.toMilliSeconds() == intervalInMs)
            {
                return;
            }
 
            intervalMs = intervalInMs;
            interval = IceUtil::Time::milliSeconds(intervalInMs);
 
            __changeInterval();
        }
 
        /**
         * Stops the thread. Only has effect if the task has been started.
         *
         *
         */
        void
        stop()
        {
            std::unique_lock lock(dataMutex);
 
            if (running)
            {
                shutdown = true;
                lock.unlock();
                //                timer->cancel(this);
                timer->destroy();
                timer = new armarx::Timer(timer->getUseSystemTime());
            }
 
            running = false;
        }
 
        /**
         * Retrieve running state of the thread
         *
         * @return whether thread is running
         */
        bool
        isRunning() const
        {
            return running;
        }
 
        /**
         * Return the execution state of the thread function.
         *
         * @return PeriodicTask::functionExecuting
         */
        bool
        isFunctionExecuting() const
        {
            std::unique_lock lock(mutexFunctionExecuting);
            return functionExecuting;
        }
 
        /**
         * @return PeriodicTaskIceBase base pointer to the current PeriodicTask instance
         */
        const PeriodicTaskIceBase&
        getStatistics() const
        {
            const PeriodicTaskIceBase* base = dynamic_cast<const PeriodicTaskIceBase*>(this);
            return *base;
        }
 
    private:
        void
        runTimerTask() override
        {
            {
                std::unique_lock lock(mutexFunctionExecuting);
                functionExecuting = true;
            }
            IceUtil::Time idleDuration;
            {
                std::unique_lock lock(dataMutex);
                threadId = LogSender::getThreadId();
 
                idleDuration = __now() - cycleStart - cycleDuration;
                cycleStart = __now();
                lastCycleStartTime = cycleStart.toMicroSeconds();
            }
 
            // call periodic function
            try
            {
                (parent->*periodicFn)();
            }
            catch (...)
            {
                handleExceptions();
            }
 
            {
                std::unique_lock lock2(dataMutex);
                std::unique_lock lock(mutexFunctionExecuting);
                // measure execution time
                cycleDuration = __now() - cycleStart;
                lastCycleDuration = cycleDuration.toMicroSeconds();
                __checkTimeslotUsage();
                __getCPULoad(idleDuration);
                __reschedule();
                functionExecuting = false;
            }
        }
 
        void
        __changeInterval()
        {
            std::unique_lock lock(mutexFunctionExecuting);
            startTime = __now().toMicroSeconds();
            executionCounter = 0;
 
            if (functionExecuting)
            {
                return;
            }
 
            timer->cancel(this);
 
            timer->schedule(this, interval);
            ARMARX_VERBOSE << "Execution interval changed to " << interval.toMilliSeconds()
                           << " ms." << std::endl;
        }
 
        void
        __reschedule()
        {
            if (isRunning())
            {
                IceUtil::Time newInterval;
                executionCounter++;
 
                if (assureMeanInterval)
                {
                    newInterval = IceUtil::Time::microSeconds(
                                      startTime + interval.toMicroSeconds() * executionCounter) -
                                  __now();
                }
                else
                {
                    newInterval = interval - IceUtil::Time::microSeconds(lastCycleDuration);
                }
 
                if (newInterval.toMicroSeconds() < 0)
                {
                    newInterval = IceUtil::Time::microSeconds(0);
                }
 
                if (!shutdown)
                {
                    timer->schedule(this, newInterval);
                }
            }
        }
 
        void
        __checkTimeslotUsage()
        {
            if (interval + delayWarningTolerance < cycleDuration)
            {
                timeExceeded = true;
                executionTime = cycleDuration.toMilliSeconds();
            }
 
            float logInterval = 30.0f;
 
            if (timeExceeded)
            {
                ARMARX_INFO << deactivateSpam(logInterval) << "periodic task '" << name << "' took "
                            << executionTime << " [ms] instead of the requested "
                            << interval.toMilliSeconds()
                            << " [ms] (this message is only posted every " << logInterval
                            << " seconds.)";
                timeExceeded = false;
            }
        }
 
        void
        __getCPULoad(IceUtil::Time& idleDuration)
        {
            double threadCPUUsage = -1;
            timespec usage;
            timespec st;
 
            if (clock_gettime(CLOCK_REALTIME, &st) != -1 &&
                clock_gettime(CLOCK_THREAD_CPUTIME_ID, &usage) != -1)
            {
                double current_cpu_thread_time = static_cast<double>(usage.tv_sec) +
                                                 static_cast<double>(usage.tv_nsec) / 1'000'000'000;
                double current_cpu_total_time = static_cast<double>(st.tv_sec) +
                                                static_cast<double>(st.tv_nsec) / 1'000'000'000;
 
                if (last_cpu_total_time != 0)
                {
                    threadCPUUsage = (current_cpu_thread_time - last_cpu_thread_time) /
                                     (current_cpu_total_time - last_cpu_total_time);
                }
 
                last_cpu_total_time = current_cpu_total_time;
                last_cpu_thread_time = current_cpu_thread_time;
            }
            else
            {
                // store the workload of this task the simple way
                threadCPUUsage = static_cast<double>(lastCycleDuration) /
                                 (lastCycleDuration + idleDuration.toMicroSeconds());
            }
 
            if (threadCPUUsage > 0)
            {
                workloadList.push_back(threadCPUUsage);
 
                if (workloadList.size() > 100.f * 1.5f)
                {
                    workloadList.erase(workloadList.begin(), workloadList.begin() + 50);
                }
            }
        }
 
        IceUtil::Time
        __now() const
        {
            if (timer->getUseSystemTime())
            {
                return IceUtil::Time::now();
            }
            else
            {
                return TimeUtil::GetTime();
            }
        }
 
        armarx::TimerPtr timer;
        T* parent;
        method_type periodicFn;
 
 
        IceUtil::Time interval;
        IceUtil::Time cycleStart;
        IceUtil::Time cycleDuration;
        IceUtil::Time logTime;
        IceUtil::Time delayWarningTolerance;
        IceUtil::Int64 executionTime;
        IceUtil::Int64 executionCounter;
 
        //        IceUtil::Time           scheduleTime;
        //        IceUtil::Time           scheduledInterval;
        //        IceUtil::Int64          realIntervalSum;
        //        IceUtil::Int64          scheduledIntervalSum;
        double last_cpu_total_time;
        double last_cpu_thread_time;
 
        bool timeExceeded;
        //! Avoids Drifting
        bool assureMeanInterval;
 
        mutable std::mutex mutexFunctionExecuting;
        bool functionExecuting;
        mutable std::mutex dataMutex;
 
        ThreadListPtr customThreadList;
        ThreadListPtr appThreadList; // prevent deletion of global applist
    };
 
 
} // namespace armarx