ref: 0ad782525ec14a5188a441bbc30f811576b529b7
dir: /codec/common/src/WelsThreadPool.cpp/
/*! * \copy * Copyright (c) 2009-2015, Cisco Systems * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions * are met: * * * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * * * Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in * the documentation and/or other materials provided with the * distribution. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE * COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE * POSSIBILITY OF SUCH DAMAGE. * * * \file WelsThreadPool.cpp * * \brief functions for Thread Pool * * \date 5/09/2012 Created * ************************************************************************************* */ #include "typedefs.h" #include "memory_align.h" #include "WelsThreadPool.h" namespace WelsCommon { namespace { CWelsLock& GetInitLock() { static CWelsLock *initLock = new CWelsLock; return *initLock; } } int32_t CWelsThreadPool::m_iRefCount = 0; int32_t CWelsThreadPool::m_iMaxThreadNum = DEFAULT_THREAD_NUM; CWelsThreadPool* CWelsThreadPool::m_pThreadPoolSelf = NULL; CWelsThreadPool::CWelsThreadPool() : m_cWaitedTasks (NULL), m_cIdleThreads (NULL), m_cBusyThreads (NULL) { } CWelsThreadPool::~CWelsThreadPool() { //fprintf(stdout, "CWelsThreadPool::~CWelsThreadPool: delete %x, %x, %x\n", m_cWaitedTasks, m_cIdleThreads, m_cBusyThreads); if (0 != m_iRefCount) { m_iRefCount = 0; Uninit(); } } WELS_THREAD_ERROR_CODE CWelsThreadPool::SetThreadNum (int32_t iMaxThreadNum) { CWelsAutoLock cLock (GetInitLock()); if (m_iRefCount != 0) { return WELS_THREAD_ERROR_GENERAL; } if (iMaxThreadNum <= 0) { iMaxThreadNum = 1; } m_iMaxThreadNum = iMaxThreadNum; return WELS_THREAD_ERROR_OK; } CWelsThreadPool* CWelsThreadPool::AddReference() { CWelsAutoLock cLock (GetInitLock()); if (m_pThreadPoolSelf == NULL) { m_pThreadPoolSelf = new CWelsThreadPool(); if (!m_pThreadPoolSelf) { return NULL; } } if (m_iRefCount == 0) { if (WELS_THREAD_ERROR_OK != m_pThreadPoolSelf->Init()) { m_pThreadPoolSelf->Uninit(); delete m_pThreadPoolSelf; m_pThreadPoolSelf = NULL; return NULL; } } ////fprintf(stdout, "m_iRefCount=%d, iMaxThreadNum=%d\n", m_iRefCount, m_iMaxThreadNum); ++ m_iRefCount; //fprintf(stdout, "m_iRefCount2=%d\n", m_iRefCount); return m_pThreadPoolSelf; } void CWelsThreadPool::RemoveInstance() { CWelsAutoLock cLock (GetInitLock()); //fprintf(stdout, "m_iRefCount=%d\n", m_iRefCount); -- m_iRefCount; if (0 == m_iRefCount) { StopAllRunning(); Uninit(); if (m_pThreadPoolSelf) { delete m_pThreadPoolSelf; m_pThreadPoolSelf = NULL; } //fprintf(stdout, "m_iRefCount=%d, IdleThreadNum=%d, BusyThreadNum=%d, WaitedTask=%d\n", m_iRefCount, GetIdleThreadNum(), GetBusyThreadNum(), GetWaitedTaskNum()); } } bool CWelsThreadPool::IsReferenced() { CWelsAutoLock cLock (GetInitLock()); return (m_iRefCount > 0); } WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStart (CWelsTaskThread* pThread, IWelsTask* pTask) { AddThreadToBusyList (pThread); //fprintf(stdout, "CWelsThreadPool::AddThreadToBusyList: Task %x at Thread %x\n", pTask, pThread); return WELS_THREAD_ERROR_OK; } WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStop (CWelsTaskThread* pThread, IWelsTask* pTask) { //fprintf(stdout, "CWelsThreadPool::OnTaskStop 0: Task %x at Thread %x Finished\n", pTask, pThread); RemoveThreadFromBusyList (pThread); AddThreadToIdleQueue (pThread); if (pTask && pTask->GetSink()) { //fprintf(stdout, "CWelsThreadPool::OnTaskStop 1: Task %x at Thread %x Finished, m_pSink=%x\n", pTask, pThread, pTask->GetSink()); pTask->GetSink()->OnTaskExecuted(); ////fprintf(stdout, "CWelsThreadPool::OnTaskStop 1: Task %x at Thread %x Finished, m_pSink=%x\n", pTask, pThread, pTask->GetSink()); } //if (m_pSink) { // m_pSink->OnTaskExecuted (pTask); //} //fprintf(stdout, "CWelsThreadPool::OnTaskStop 2: Task %x at Thread %x Finished\n", pTask, pThread); SignalThread(); //fprintf(stdout, "ThreadPool: Task %x at Thread %x Finished\n", pTask, pThread); return WELS_THREAD_ERROR_OK; } WELS_THREAD_ERROR_CODE CWelsThreadPool::Init() { //fprintf(stdout, "Enter WelsThreadPool Init\n"); CWelsAutoLock cLock (m_cLockPool); m_cWaitedTasks = new CWelsNonDuplicatedList<IWelsTask>(); m_cIdleThreads = new CWelsNonDuplicatedList<CWelsTaskThread>(); m_cBusyThreads = new CWelsList<CWelsTaskThread>(); if (NULL == m_cWaitedTasks || NULL == m_cIdleThreads || NULL == m_cBusyThreads) { return WELS_THREAD_ERROR_GENERAL; } for (int32_t i = 0; i < m_iMaxThreadNum; i++) { if (WELS_THREAD_ERROR_OK != CreateIdleThread()) { return WELS_THREAD_ERROR_GENERAL; } } if (WELS_THREAD_ERROR_OK != Start()) { return WELS_THREAD_ERROR_GENERAL; } return WELS_THREAD_ERROR_OK; } WELS_THREAD_ERROR_CODE CWelsThreadPool::StopAllRunning() { WELS_THREAD_ERROR_CODE iReturn = WELS_THREAD_ERROR_OK; ClearWaitedTasks(); while (GetBusyThreadNum() > 0) { //WELS_INFO_TRACE ("CWelsThreadPool::Uninit - Waiting all thread to exit"); WelsSleep (10); } if (GetIdleThreadNum() != m_iMaxThreadNum) { iReturn = WELS_THREAD_ERROR_GENERAL; } return iReturn; } WELS_THREAD_ERROR_CODE CWelsThreadPool::Uninit() { WELS_THREAD_ERROR_CODE iReturn = WELS_THREAD_ERROR_OK; CWelsAutoLock cLock (m_cLockPool); iReturn = StopAllRunning(); if (WELS_THREAD_ERROR_OK != iReturn) { return iReturn; } m_cLockIdleTasks.Lock(); while (m_cIdleThreads->size() > 0) { DestroyThread (m_cIdleThreads->begin()); m_cIdleThreads->pop_front(); } m_cLockIdleTasks.Unlock(); Kill(); WELS_DELETE_OP (m_cWaitedTasks); WELS_DELETE_OP (m_cIdleThreads); WELS_DELETE_OP (m_cBusyThreads); return iReturn; } void CWelsThreadPool::ExecuteTask() { //fprintf(stdout, "ThreadPool: scheduled tasks: ExecuteTask\n"); CWelsTaskThread* pThread = NULL; IWelsTask* pTask = NULL; while (GetWaitedTaskNum() > 0) { //fprintf(stdout, "ThreadPool: ExecuteTask: waiting task %d\n", GetWaitedTaskNum()); pThread = GetIdleThread(); if (pThread == NULL) { //fprintf(stdout, "ThreadPool: ExecuteTask: no IdleThread\n"); break; } pTask = GetWaitedTask(); //fprintf(stdout, "ThreadPool: ExecuteTask = %x at thread %x\n", pTask, pThread); if (pTask) { pThread->SetTask (pTask); } else { AddThreadToIdleQueue (pThread); } } } WELS_THREAD_ERROR_CODE CWelsThreadPool::QueueTask (IWelsTask* pTask) { CWelsAutoLock cLock (m_cLockPool); //fprintf(stdout, "CWelsThreadPool::QueueTask: %d, pTask=%x\n", m_iRefCount, pTask); if (GetWaitedTaskNum() == 0) { CWelsTaskThread* pThread = GetIdleThread(); if (pThread != NULL) { //fprintf(stdout, "ThreadPool: ExecuteTask = %x at thread %x\n", pTask, pThread); pThread->SetTask (pTask); return WELS_THREAD_ERROR_OK; } } //fprintf(stdout, "ThreadPool: AddTaskToWaitedList: %x\n", pTask); if (false == AddTaskToWaitedList (pTask)) { return WELS_THREAD_ERROR_GENERAL; } //fprintf(stdout, "ThreadPool: SignalThread: %x\n", pTask); SignalThread(); return WELS_THREAD_ERROR_OK; } WELS_THREAD_ERROR_CODE CWelsThreadPool::CreateIdleThread() { CWelsTaskThread* pThread = new CWelsTaskThread (this); if (NULL == pThread) { return WELS_THREAD_ERROR_GENERAL; } if (WELS_THREAD_ERROR_OK != pThread->Start()) { return WELS_THREAD_ERROR_GENERAL; } //fprintf(stdout, "ThreadPool: AddThreadToIdleQueue: %x\n", pThread); AddThreadToIdleQueue (pThread); return WELS_THREAD_ERROR_OK; } void CWelsThreadPool::DestroyThread (CWelsTaskThread* pThread) { pThread->Kill(); WELS_DELETE_OP (pThread); return; } WELS_THREAD_ERROR_CODE CWelsThreadPool::AddThreadToIdleQueue (CWelsTaskThread* pThread) { CWelsAutoLock cLock (m_cLockIdleTasks); m_cIdleThreads->push_back (pThread); return WELS_THREAD_ERROR_OK; } WELS_THREAD_ERROR_CODE CWelsThreadPool::AddThreadToBusyList (CWelsTaskThread* pThread) { CWelsAutoLock cLock (m_cLockBusyTasks); m_cBusyThreads->push_back (pThread); return WELS_THREAD_ERROR_OK; } WELS_THREAD_ERROR_CODE CWelsThreadPool::RemoveThreadFromBusyList (CWelsTaskThread* pThread) { CWelsAutoLock cLock (m_cLockBusyTasks); if (m_cBusyThreads->erase (pThread)) { return WELS_THREAD_ERROR_OK; } else { return WELS_THREAD_ERROR_GENERAL; } } bool CWelsThreadPool::AddTaskToWaitedList (IWelsTask* pTask) { CWelsAutoLock cLock (m_cLockWaitedTasks); return m_cWaitedTasks->push_back (pTask); } CWelsTaskThread* CWelsThreadPool::GetIdleThread() { CWelsAutoLock cLock (m_cLockIdleTasks); if (NULL == m_cIdleThreads || m_cIdleThreads->size() == 0) { return NULL; } //fprintf(stdout, "CWelsThreadPool::GetIdleThread=%d\n", m_cIdleThreads->size()); CWelsTaskThread* pThread = m_cIdleThreads->begin(); m_cIdleThreads->pop_front(); return pThread; } int32_t CWelsThreadPool::GetBusyThreadNum() { return (m_cBusyThreads?m_cBusyThreads->size():0); } int32_t CWelsThreadPool::GetIdleThreadNum() { return (m_cIdleThreads?m_cIdleThreads->size():0); } int32_t CWelsThreadPool::GetWaitedTaskNum() { return (m_cWaitedTasks?m_cWaitedTasks->size():0); } IWelsTask* CWelsThreadPool::GetWaitedTask() { CWelsAutoLock lock (m_cLockWaitedTasks); if (NULL==m_cWaitedTasks || m_cWaitedTasks->size() == 0) { return NULL; } IWelsTask* pTask = m_cWaitedTasks->begin(); m_cWaitedTasks->pop_front(); return pTask; } void CWelsThreadPool::ClearWaitedTasks() { CWelsAutoLock cLock (m_cLockWaitedTasks); if (NULL == m_cWaitedTasks) { return; } IWelsTask* pTask = NULL; while (0 != m_cWaitedTasks->size()) { pTask = m_cWaitedTasks->begin(); if (pTask->GetSink()) { pTask->GetSink()->OnTaskCancelled(); } m_cWaitedTasks->pop_front(); } } }