/*!

 * \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();

  assert (0 == GetBusyThreadNum());

  m_cLockIdleTasks.Lock();

  while (m_cIdleThreads && 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()) {

    WELS_DELETE_OP (pThread);

    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();

  }

}

}