// CONSTRUCTOR
//------------------------------------------------------------------------------
JobQueue::JobQueue( uint32_t numWorkerThreads ) :
	m_NumLocalJobsActive( 0 ),
	m_DistributableAvailableJobs( 1024, true ),
	m_DistributableJobsMemoryUsage( 0 ),
	m_DistributedJobsRemote( 1204, true ),
	m_DistributedJobsLocal( 128, true ),
	m_DistributedJobsCancelled( 128, true ),
	m_CompletedJobs( 1024, true ),
	m_CompletedJobsFailed( 1024, true ),
	m_CompletedJobs2( 1024, true ),
	m_CompletedJobsFailed2( 1024, true ),
	m_Workers( numWorkerThreads, false )
{
	WorkerThread::InitTmpDir();

	for ( uint32_t i=0; i<numWorkerThreads; ++i )
	{
		// identify each worker with an id starting from 1
		// (the "main" thread is considered 0)
		uint32_t threadIndex = ( i + 1 );
		WorkerThread * wt = FNEW( WorkerThread( threadIndex ) );
		wt->Init();
		m_Workers.Append( wt );
	}
}

ThreadPool::~ThreadPool( )
{
    // --------------------------------------------------------------
    // Request Termination of all worker threads.
    // --------------------------------------------------------------

    m_mList.lock();
    WorkerThreadList list = m_lstThreads;
    m_mList.unlock();

    WorkerThreadList::iterator it = list.begin(); 
    for (; it != list.end(); ++it)
    {
        if (*it != NULL) 
            (*it)->RequestTerminate();
    }

    while (!list.empty())
    {
        WorkerThread *pThread = list.front();
        list.pop_front();
        if (pThread != NULL) 
        {
            pThread->wait();
            delete pThread;
        }
    }
}

void Controller::startWorkInAThread()
{
    WorkerThread *workerThread = new WorkerThread(this);
    connect(workerThread, &WorkerThread::resultReady, this, &Controller::handleResults);
    connect(workerThread, &WorkerThread::finished, workerThread, &QObject::deleteLater);
    workerThread->start();
}

/*!
 * \brief WorkerThreadPool::findResponsiveThread
 * \return
 */
WorkerThread *WorkerThreadPool::findResponsiveThread()
{
    QMutexLocker l(&threadListMutex);

    if(idleThreads.size()) {
        WorkerThread* wt = idleThreads.takeFirst();
        qDebug()<<"[Main] Using Thread "<<wt->threadId()<<" from idle List";
        return wt;
    } else {
        qDebug()<<"[Main] No Idle Threads trying to find a responsive one";

        int myThread = -1;
        for(int i = 0; i < threads.size(); i++) {
            WorkerRunnable::Load load = threads.at(i)->workLoad();
            if(load == WorkerRunnable::Low) {
                myThread = i;
                break;
            } else if(load == WorkerRunnable::Medium && myThread == -1) {
                myThread = i;
                continue;
            }
        }

        if(myThread >= 0) {
            //move the thread to the back, so it will be picked later the next time
            WorkerThread* t = threads.takeAt(myThread);
            threads.append(t);

            return t;
        }

        //we did not find a thread , enqueue it
    }
    return 0;
}

/** \brief The normal thread run function.
 *
 * The function to be started by normal workers.
 */
static void run_normal()
{
  boost::mutex::scoped_lock scope(mut);
  boost::thread::id tid = boost::this_thread::get_id();
  WorkerThread* thr = (*threads.find(tid)).second.get();

  while(!quitting)
  {
    if(inner_run_important(scope))
    {
      continue;
    }
    if(inner_run_normal(scope))
    {
      continue;
    }
    wake_normal();
    workers_active.remove(thr);
    workers_sleeping.addLast(thr);
    thr->suspend(scope);
  }

  // Remove from active workers before exiting.
  workers_active.remove(thr);
}

int main(int argc, char **args)
{
	WorkerThread wt;
	wt.start();
	wt.wait();
	return 0;
}

unsigned WINAPI WorkerThread::ThreadMain(LPVOID thread){

	WorkerThread* workerThread = (WorkerThread*)thread;
	workerThread->run();
	return 0;

}

/**
 * Called by the monitor thread to harvest idle threads.
 */
void 
ThreadPool::harvestSpareWorkers() {

	PoolObject *po;
	int toFree = 0;

	cond.lock();
	if (stopThePool) {
		cond.unlock();
		return;
	}
	if ((currentThreadCount - currentThreadsBusy) > maxSpareThreads) {
		toFree = currentThreadCount -
			currentThreadsBusy -
			maxSpareThreads;

		if (debug) 
			fprintf(stdout, "ThreadPool.harvestSpareWorkers: Freeing %d threads because they are idle\n",
				toFree);
		for (int i = 0 ; i < toFree ; i++) {
			po = (PoolObject *) idleThreads.pop();
			WorkerThread *c = po->getThread();
			c->terminate();
			currentThreadCount --;
		}
		if (debug)
			fprintf(stdout, "ThreadPool.harvestSpareWorkers: Current Thread Count = %d\n",
				currentThreadCount);
	}
	cond.unlock();
}

QDebug tDebug()
{
    WorkerThread* t = qobject_cast<WorkerThread*>(QThread::currentThread());
    if(t)
        return qDebug()<<"["<<t->threadId()<<"]";

    return qDebug();
}

// Called by a worker thread on return from a job.
void WorkerThread::JobCleanup(void *param)
{
    WorkerThread *worker = (WorkerThread *)param;

    (void) pthread_mutex_lock(&(worker->m_worker_mutex));
    if (worker->m_worker_flags & WORKER_WAIT)
        worker->NotifyWaiters();
}

void Task::startWorkerThread(const QString &id)
{
  L_FUNC(QString("id='%1'").arg(id));

  WorkerThread *workerThread = new WorkerThread(id, this);
  connect(workerThread, SIGNAL(resultReady(const QString&)), this, SLOT(slotResultReady(const QString&)));
  connect(workerThread, SIGNAL(finished()), workerThread, SLOT(deleteLater()));
  workerThread->start();
}

void TaskScheduler::wakeUpOneThread()
{
	WorkerThread* thread = mIdleThreads.pop();
	if(thread)
	{
		SCHEDULER_DEBUG("wakeUpOneThread -- thread #" << thread->getThreadId() << " -- #" << thread->getDoWorkEvent()->getHandle());
		thread->getDoWorkEvent()->fire();
	}
}

void *Thread_Func(void *arg)
{
    WorkerThread *worker = new WorkerThread((RequestQueue *)arg);
    signal(SIGTERM, cleanExit);
    signal(SIGINT, cleanExit);
    signal(SIGPIPE, SIG_IGN);
    worker->mainloop();
    
    pthread_exit(NULL);
}

void MainWindow::on_learnDicOkBtn_clicked()
{
	ui->learnDicOkBtn->setEnabled(false);
	ui->saveCfBtn->setEnabled(false);
	WorkerThread *workerThread = new WorkerThread();
	connect(workerThread, SIGNAL(resultReady()), this, SLOT(updateSisrButtons()));
	connect(workerThread, SIGNAL(exceptionOccurs()), this, SLOT(sisrExceptionOccurs()));

	workerThread->start();
}

// MainWrapper
//------------------------------------------------------------------------------
/*static*/ uint32_t WorkerThread::ThreadWrapperFunc( void * param )
{
	WorkerThread * wt = reinterpret_cast< WorkerThread * >( param );
	s_WorkerThreadThreadIndex = wt->m_ThreadIndex;

	CreateThreadLocalTmpDir();

	wt->Main();
	return 0;
}

static void InitiateReconnect()
{
    WorkerThread *thr = WorkerThread::This();
    if (thr)
    {
        // Defer sending the completion of exposure message until after
        // the camera re-connecttion attempt
        thr->SetSkipExposeComplete();
    }
    pFrame->TryReconnect();
}

// worker thread is a thread that wait work in the work queue
WorkerThread * WorkerThread::Create(pthread_attr_t *attr)
{
    WorkerThread *worker;
    if((worker = new (std::nothrow)WorkerThread(attr)) == NULL)
    {
        errno = ENOMEM;
        return (NULL);
    }
    worker->CreateWorker();
    return (worker);
}

	void WorkerThread::ConnEventCB(bufferevent *bev, short int events,
	        void * ctx)
	{
		ConnItem * pitem = static_cast<ConnItem *>(ctx);
		WorkerThread * pwt = static_cast<WorkerThread*>(pitem->pthis);
		if (pitem->log_fd != -1)	//连接意外关闭，关闭写日志文件描述符
			close(pitem->log_fd);
		pwt->DeleteConnItem(*pitem);	//从线程队列中删除连接对象
		bufferevent_free(bev);
		msghandler->sessionClosed(pitem);
	}

bool TaskScheduler::allThreadsIdle()
{
	// ugly, TODO
	uint16 idleCount = 0;
	for(uint16 i = 0; i < mThreadsCount; i++)
	{
		WorkerThread* thread = mAllThreads[i];
		if(thread->isIdle()) idleCount++;
	}

	return idleCount == mThreadsCount;
}

TEST(WorkerThread, FutureVoid) {
  int callbacks = 0;
  WorkerThread<void(int)> worker;
  future<void> f = worker.QueueWork(
      [&](int v) {
        ++callbacks;
        ASSERT_EQ(v, 3);
      },
      3);
  f.get();
  ASSERT_EQ(callbacks, 1);
}