AampScheduler.cpp

Go to the documentation of this file.

/*
 * If not stated otherwise in this file or this component's license file the
 * following copyright and licenses apply:
 *
 * Copyright 2020 RDK Management
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
*/
 
/**
 * @file AampScheduler.cpp
 * @brief Class to schedule commands for async execution
 */
 
#include "AampScheduler.h"
 
/**
 * @brief AampScheduler Constructor
 */
AampScheduler::AampScheduler() : mTaskQueue(), mQMutex(), mQCond(),
        mSchedulerRunning(false), mSchedulerThread(), mExMutex(),
        mExLock(mExMutex, std::defer_lock), mNextTaskId(AAMP_SCHEDULER_ID_DEFAULT),
        mCurrentTaskId(AAMP_TASK_ID_INVALID), mLockOut(false),
        mLogObj(NULL),mState(eSTATE_IDLE)
{
}
 
/**
 * @brief AampScheduler Destructor
 */
AampScheduler::~AampScheduler()
{
        if (mSchedulerRunning)
        {
                StopScheduler();
        }
}
 
/**
 * @brief To start scheduler thread
 */
void AampScheduler::StartScheduler()
{
        //Turn on thread for processing async operations
        std::lock_guard<std::mutex>lock(mQMutex);
        mSchedulerThread = std::thread(std::bind(&AampScheduler::ExecuteAsyncTask, this));
        mSchedulerRunning = true;
        AAMPLOG_WARN("Started Async Worker Thread");
}
 
/**
 * @brief To schedule a task to be executed later
 */
int AampScheduler::ScheduleTask(AsyncTaskObj obj)
{
        int id = AAMP_TASK_ID_INVALID;
        if (mSchedulerRunning)
        {
 
                if( mState == eSTATE_ERROR || mState == eSTATE_RELEASED)
                        return id;
 
                std::lock_guard<std::mutex>lock(mQMutex);
                if (!mLockOut)
                {
                        id = mNextTaskId++;
                        // Upper limit check
                        if (mNextTaskId >= AAMP_SCHEDULER_ID_MAX_VALUE)
                        {
                                mNextTaskId = AAMP_SCHEDULER_ID_DEFAULT;
                        }
                        obj.mId = id;
                        mTaskQueue.push_back(obj);
                        mQCond.notify_one();
                }
                else
                {
                        // Operation is skipped here, this might happen due to race conditions during normal operation, hence setting as info log
                        AAMPLOG_INFO("Warning: Attempting to schedule a task when scheduler is locked out, skipping operation %s!!", obj.mTaskName.c_str());
                }
        }
        else
        {
                AAMPLOG_ERR("Attempting to schedule a task when scheduler is not running, undefined behavior, task ignored:%s",obj.mTaskName.c_str());
        }
        return id;
}
 
/**
 * @brief Executes scheduled tasks - invoked by thread
 */
void AampScheduler::ExecuteAsyncTask()
{
        std::unique_lock<std::mutex>queueLock(mQMutex);
        while (mSchedulerRunning)
        {
                if (mTaskQueue.empty())
                {
                        mQCond.wait(queueLock);
                }
                else
                {
                        /* DELIA-57121
                        Take the execution lock before taking a task from the queue
                        otherwise this function could hold a task, out of the queue,
                        that cannot be deleted by RemoveAllTasks()!
                        Allow the queue to be modified while waiting.*/
                        queueLock.unlock();
                        std::lock_guard<std::mutex>executionLock(mExMutex);
                        queueLock.lock();
 
                        //DELIA-57121 - note: mTaskQueue could have been modified while waiting for execute permission
                        if (!mTaskQueue.empty())
                        {
                                AsyncTaskObj obj = mTaskQueue.front();
                                mTaskQueue.pop_front();
                                if (obj.mId != AAMP_TASK_ID_INVALID)
                                {
                                        mCurrentTaskId = obj.mId;
                                        AAMPLOG_INFO("Found entry in function queue!!, task:%s. State:%d",obj.mTaskName.c_str(),mState);
                                        if( mState != eSTATE_ERROR && mState != eSTATE_RELEASED)
                                        {
                                                //Unlock so that new entries can be added to queue while function executes
                                                queueLock.unlock();
 
                                                AAMPLOG_WARN("SchedulerTask Execution:%s",obj.mTaskName.c_str());
                                                //Execute function
                                                obj.mTask(obj.mData);
                                                //May be used in a wait() in future loops, it needs to be locked
                                                queueLock.lock();
                                        }
                                }
                                else
                                {
                                        AAMPLOG_ERR("Scheduler found a task with invalid ID, skip task!");
                                }
                        }
                }
        }
        AAMPLOG_INFO("Exited Async Worker Thread");
}
 
/**
 * @brief To remove all scheduled tasks and prevent further tasks from scheduling
 */
void AampScheduler::RemoveAllTasks()
{
        std::lock_guard<std::mutex>lock(mQMutex);
        if(!mLockOut)
        {
                AAMPLOG_WARN("The scheduler is active.  An active task may continue to execute after this function exits.  Call SuspendScheduler() prior to this function to prevent this.");
        }
        if (!mTaskQueue.empty())
        {
                AAMPLOG_WARN("Clearing up %zu entries from mFuncQueue", mTaskQueue.size());
                mTaskQueue.clear();
        }
}
 
/**
 * @brief To stop scheduler and associated resources
 */
void AampScheduler::StopScheduler()
{
        AAMPLOG_WARN("Stopping Async Worker Thread");
        // Clean up things in queue
        mSchedulerRunning = false;
 
        //DELIA-57121 allow StopScheduler() to be called without warning from a nonsuspended state and
        //DELIA-57122 not cause an error in ResumeScheduler() below due to trying to unlock an unlocked lock
        if(!mLockOut)
        {
                SuspendScheduler();
        }
 
        RemoveAllTasks();
 
        //DELIA-57122 prevent possible deadlock where mSchedulerThread is waiting for mExLock/mExMutex
        ResumeScheduler();
        mQCond.notify_one();
    if (mSchedulerThread.joinable())
        mSchedulerThread.join();
}
 
/**
 * @brief To acquire execution lock for synchronisation purposes
 */
void AampScheduler::SuspendScheduler()
{
        mExLock.lock();
        std::lock_guard<std::mutex>lock(mQMutex);
        mLockOut = true;
}
 
/**
 * @brief To release execution lock
 */
void AampScheduler::ResumeScheduler()
{
        mExLock.unlock();
        std::lock_guard<std::mutex>lock(mQMutex);
        mLockOut = false;
}
 
/**
 * @brief To remove a scheduled tasks with ID
 */
bool AampScheduler::RemoveTask(int id)
{
        bool ret = false;
        std::lock_guard<std::mutex>lock(mQMutex);
        // Make sure its not currently executing/executed task
        if (id != AAMP_TASK_ID_INVALID && mCurrentTaskId != id)
        {
                for (auto it = mTaskQueue.begin(); it != mTaskQueue.end(); )
                {
                        if (it->mId == id)
                        {
                                mTaskQueue.erase(it);
                                ret = true;
                                break;
                        }
                        else
                        {
                                it++;
                        }
                }
        }
        return ret;
}
 
/**
 * @brief To enable scheduler to queue new tasks
 */
void AampScheduler::EnableScheduleTask()
{
        std::lock_guard<std::mutex>lock(mQMutex);
        mLockOut = false;
}
 
/**
 * @brief To player state to Scheduler
 */
void AampScheduler::SetState(PrivAAMPState sstate)
{
        mState = sstate;
}
 
#ifdef __UNIT_TESTING__
 
 
void printTask(const char *str)
{
    AAMPLOG_WARN("Inside task function :%s\n",str);
    int seconds = rand() % 10 + 1;
    sleep(seconds);
}
 
 
void AddTask(AampScheduler &mScheduler, std::string str )
{
    mScheduler.ScheduleTask(AsyncTaskObj([str](void *data)
                                        {
                                                printTask(str.c_str());
                                        }, (void *) nullptr, __FUNCTION__));
 
}
 
 
void task1(void *inst) {
    std::string input="Task1";
    AampScheduler *mScheduler = (AampScheduler *)inst;
    for(int i=0;i<10;i++)
    {
        mScheduler->ScheduleTask(AsyncTaskObj([input](void *data)
                                        {
                                                printTask(input.c_str());
                                        }, (void *) nullptr, __FUNCTION__));
 
        int state = rand() % 13 + 1;
        mScheduler->SetState((PrivAAMPState)state);
        AAMPLOG_WARN("setting state:%d",state);
    }
}
 
void task2(void *inst) {
    std::string input="Task2";
    AampScheduler *mScheduler = (AampScheduler *)inst;
    mScheduler->ScheduleTask(AsyncTaskObj([input](void *data)
                                        {
                                                printTask(input.c_str());
                                        }, (void *) nullptr, __FUNCTION__));
 
}
 
int main()
{
    AAMPLOG_WARN("Testing the Player Scheduler Class\n");
    AampScheduler mScheduler;
    srand(time(0));
    //Test 1: Simple start / stop scheduler
    //mScheduler.StartScheduler();
    //mScheduler.StopScheduler();
    // Test 2: start but no stop
    //mScheduler.StartScheduler();
    // Test 3 :
    //mScheduler.StartScheduler();
    //AddTask(mScheduler, "Test1");
    //sleep(10);
    //mScheduler.StopScheduler();
    // Test 4 :
    //mScheduler.StartScheduler();
    //mScheduler.SetState(eSTATE_RELEASED);
    //AddTask(mScheduler, "Test2");
    //sleep(10);
    //mScheduler.StopScheduler();
    // Test 5 :
    //mScheduler.StartScheduler();
    //AddTask(mScheduler, "Test2");
    //mScheduler.SetState(eSTATE_RELEASED);
    //sleep(10);
    //mScheduler.StopScheduler();
    // Test 6: Stop without start
    //mScheduler.StopScheduler();
    // Test 7: Other API calls with start Scheduler
    //for(int i=0;i<10;i++)
    //{
    //    AddTask(mScheduler, "LoopTest");
    //    mScheduler.RemoveAllTasks();
    //    mScheduler.SuspendScheduler();
    //    mScheduler.ResumeScheduler();
    //}
    // Test 8 :
    mScheduler.StartScheduler();
    std::thread t1(task1,(void *)&mScheduler);
    std::thread t2(task2,(void *)&mScheduler);
    sleep(10);
    t1.join();
    t2.join();
    mScheduler.StopScheduler();
    return 0;
}
#endif