RunningTask.cpp

Go to the documentation of this file.

#include "RunningTask.h"
 
#include "ThreadList.h"
 
#include <ArmarXCore/core/application/Application.h>
#include <ArmarXCore/core/exceptions/LocalException.h>
 
#include <condition_variable>
#include <mutex>
 
namespace armarx
{
 
    struct RunningTaskBase::Impl
    {
        //! mutex for the stop function
        std::mutex stoppingMutex;
        //! mutex for the stopped status
        std::mutex stopMutex;
        std::condition_variable stopCondition;
        //! mutex for the finished status
        std::mutex finishedMutex;
        std::condition_variable finishedCondition;
 
        ThreadListPtr customThreadList;
        bool threadJoined = false;
    };
 
    RunningTaskBase::RunningTaskBase(std::string const& name)
        : impl(new Impl)
    {
        this->running = false;
        this->stopped = false;
        this->finished = false;
        this->workload = 0.0f;
 
        setName(name);
    }
 
    RunningTaskBase::~RunningTaskBase()
    {
        stop();
    }
 
    void RunningTaskBase::setName(const std::string& name)
    {
        RunningTaskIceBase::name = name;
    }
 
    void RunningTaskBase::setThreadList(ThreadListPtr threadList)
    {
        impl->customThreadList = threadList;
 
        if (isRunning())
        {
            impl->customThreadList->addRunningTask(this);
        }
    }
 
    void RunningTaskBase::start()
    {
        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 && !finished)
        {
            impl->threadJoined = false;
            startTime = IceUtil::Time::now().toMicroSeconds();
            IceUtil::Thread::start();
 
            running = true;
        }
 
        if (!running && finished)
        {
            throw LocalException("Running Task '" +  RunningTaskIceBase::name  + "' is already finished and cannot be started again.");
        }
    }
 
    void RunningTaskBase::stop(bool waitForJoin)
    {
        std::unique_lock stoppingMutexLock{impl->stoppingMutex, std::defer_lock};
 
        if (waitForJoin)
        {
            impl->stoppingMutex.lock();    // thread is beeing stopped at the moment
        }
        else if (!stoppingMutexLock.try_lock())
        {
            return;    // thread is already being stopped
        }
 
        try
        {
            if (running)
            {
                if (!stopped)
                {
                    {
                        std::unique_lock lock(impl->stopMutex);
                        stopped = true;
                        impl->stopCondition.notify_all();
                    }
                }
            }
 
            if (waitForJoin && !impl->threadJoined)
            {
                impl->threadJoined = true;
                getThreadControl().join();
            }
 
            impl->stoppingMutex.unlock();
 
        }
        catch (IceUtil::ThreadSyscallException& e) // happens for example when the thread is already erased
        {
            impl->stoppingMutex.unlock();
        }
        catch (IceUtil::ThreadNotStartedException& e)
        {
            impl->stoppingMutex.unlock();
        }
        catch (...) // make sure that the mutex is unlocked on all exceptions
        {
            ARMARX_INFO_S << "Got exception in RunningTask::stop()";
            impl->stoppingMutex.unlock();
            throw;
        }
 
    }
 
    void RunningTaskBase::join()
    {
        std::unique_lock lock(impl->stoppingMutex);
        if (!impl->threadJoined)
        {
            impl->threadJoined = true;
            getThreadControl().join();
        }
    }
 
    bool RunningTaskBase::isRunning() const
    {
        return this->running;
    }
 
    bool RunningTaskBase::isFinished() const
    {
        return this->finished;
    }
 
    bool RunningTaskBase::waitForFinished(int timeoutMS)
    {
        std::unique_lock lock(impl->finishedMutex);
        if (!running && !finished) // the task never started
        {
            return true;
        }
        while (!isFinished())
        {
            if (timeoutMS == -1)
            {
                impl->finishedCondition.wait(lock);
            }
            else if (impl->finishedCondition.wait_for(lock, std::chrono::milliseconds(timeoutMS)) == std::cv_status::timeout)
            {
                return false;
            }
 
        }
        return true;
    }
 
    bool RunningTaskBase::isStopped()
    {
        lastFeedbackTime = IceUtil::Time::now().toMicroSeconds();
        return stopped;
    }
 
    void RunningTaskBase::waitForStop()
    {
        std::unique_lock lock(impl->stopMutex);
 
        while (!isStopped())
        {
            impl->stopCondition.wait(lock);
        }
 
    }
 
    std::string RunningTaskBase::getName() const
    {
        return RunningTaskIceBase::name;
    }
 
    void RunningTaskBase::run()
    {
        threadId = LogSender::getThreadId();
        ThreadList::getApplicationThreadList()->addRunningTask(this);
 
        if (impl->customThreadList)
        {
            impl->customThreadList->addRunningTask(this);
        }
 
        try
        {
            if (callback)
            {
                callback();
            }
            else
            {
                throw std::runtime_error("No callback defined in RunningTask");
            }
        }
        catch (...)
        {
            handleExceptions();
        }
 
        std::unique_lock lock(impl->finishedMutex);
        finished = true;
        running = false;
        ThreadList::getApplicationThreadList()->removeRunningTask(this);
 
        if (impl->customThreadList)
        {
            impl->customThreadList->removeRunningTask(this);
        }
 
        impl->stopCondition.notify_all();
        impl->finishedCondition.notify_all();
    }
 
 
}