taskqueue.cpp

Go to the documentation of this file.

// Program Name: taskqueue.cpp
// Created     : Oct. 24, 2005
//
// Purpose     : Used to process delayed tasks
//                                                                            
// Copyright (c) 2005 David Blain <dblain@mythtv.org>
//                                          
// Licensed under the GPL v2 or later, see COPYING for details                    
//

#include "taskqueue.h"
#include "mythlogging.h"

#include <QDateTime>

#include <iostream>

// Define Global instance 

static QMutex g_pTaskQueueCreationLock;
TaskQueue* TaskQueue::g_pTaskQueue = NULL;

// 
// Task Implementation
//

long Task::m_nTaskCount = 0;

//

Task::Task()
{
    m_nTaskId = m_nTaskCount++;
}

//

Task::~Task()
{
}

//
// Task Queue Implementation
//

//

TaskQueue* TaskQueue::Instance()
{
    QMutexLocker locker(&g_pTaskQueueCreationLock);
    return g_pTaskQueue ? g_pTaskQueue : (g_pTaskQueue = new TaskQueue());
}

//

void TaskQueue::Shutdown()
{
    QMutexLocker locker(&g_pTaskQueueCreationLock);
    delete g_pTaskQueue;
    g_pTaskQueue = NULL;
}

//

TaskQueue::TaskQueue() : MThread("TaskQueue"), m_bTermRequested( false )
{
    LOG(VB_UPNP, LOG_INFO, "Starting TaskQueue Thread...");

    start();

    LOG(VB_UPNP, LOG_INFO, "TaskQueue Thread Started.");
}

//

TaskQueue::~TaskQueue()
{
    m_bTermRequested = true; 

    wait();

    Clear();
}

void TaskQueue::RequestTerminate()
{
    m_bTermRequested = true;
}

//

void TaskQueue::run( )
{
    RunProlog();

    Task *pTask;

    LOG(VB_UPNP, LOG_INFO, "TaskQueue Thread Running.");

    while ( !m_bTermRequested )
    {
        // ------------------------------------------------------------------
        // Process Any Tasks that may need to be executed.
        // ------------------------------------------------------------------

        TaskTime ttNow;
        gettimeofday( (&ttNow), NULL );

        if ((pTask = GetNextExpiredTask( ttNow )) != NULL)
        {
            try
            {
                pTask->Execute( this );
                pTask->Release();
            }
            catch( ... )
            {
                LOG(VB_GENERAL, LOG_ERR, "Call to Execute threw an exception.");
            }

        }
        // Make sure to throttle our processing.

        msleep( 100 );
    }

    RunEpilog();
}

//

void TaskQueue::Clear( )
{
    m_mutex.lock(); 

    for ( TaskMap::iterator it  = m_mapTasks.begin();
                            it != m_mapTasks.end();
                          ++it )
    {
        if ((*it).second != NULL)
            (*it).second->Release();
    }

    m_mapTasks.clear();

    m_mutex.unlock(); 
}

//

void TaskQueue::AddTask( long msec, Task *pTask )
{
    TaskTime tt;
    gettimeofday( (&tt), NULL );

    AddMicroSecToTaskTime( tt, (msec * 1000) );

    AddTask( tt, pTask );
}

//

void TaskQueue::AddTask( TaskTime ttKey, Task *pTask )
{

    if (pTask != NULL)
    {
        m_mutex.lock(); 
        pTask->AddRef();
        m_mapTasks.insert( TaskMap::value_type( ttKey, pTask ));
        m_mutex.unlock();
    }
}

//

void TaskQueue::AddTask( Task *pTask )
{

    if (pTask != NULL)
    {
        TaskTime tt;
        gettimeofday( (&tt), NULL );

        AddTask( tt, pTask );
    }
}

//

Task *TaskQueue::GetNextExpiredTask( TaskTime tt, long nWithinMilliSecs /*=50*/ )
{
    Task *pTask = NULL;

    AddMicroSecToTaskTime( tt, nWithinMilliSecs * 1000 );

    m_mutex.lock(); 

    TaskMap::iterator it = m_mapTasks.begin();

    if (it != m_mapTasks.end())
    {
        TaskTime ttTask = (*it).first;

        if (ttTask < tt)
        {
            // Do not release here... caller must call release.

            pTask = (*it).second;
        
            m_mapTasks.erase( it );    
        }
    }
    m_mutex.unlock(); 

    return pTask;
}

---