///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoAdminThreadPool.cc  
// ----------------------                                                     
// Cego admin threadpool class implementation
//                                                         
// Design and Implementation by Bjoern Lemke
//               
// (C)opyright 2000-2010 Bjoern Lemke
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2, or (at your option)
// any later version.
// 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; see the file COPYING.  If not, write to
// the Free Software Foundation, 59 Temple Place - Suite 330,
// Boston, MA 02111-1307, USA.
//
// IMPLEMENTATION MODULE
//
// Class: CegoAdminThreadPool
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

#include "CegoAdminThreadPool.h"

#include <lfc/Sleeper.h>
#include <lfc/NetHandler.h>
#include <lfc/Net.h>
#include <lfc/Exception.h>
#include <lfc/ThreadLock.h>
#include <lfc/XMLSuite.h>

#include "CegoDefs.h"
#include "CegoAdminThread.h"

#include <stdlib.h>
#include <string.h>


static ThreadLock queueLock("ADMQUEUE"); 
extern bool __lockStatOn;

CegoAdminThreadPool::CegoAdminThreadPool() : Thread()
{
    queueLock.init(LCKMNG_LOCKWAITDELAY);
}

CegoAdminThreadPool::CegoAdminThreadPool(int poolLimit, CegoDatabaseManager *pDBMng, CegoDbThreadPool *pDbPool, CegoLogThreadPool *pLogPool) : Thread()
{

    queueLock.init(LCKMNG_LOCKWAITDELAY, __lockStatOn);
    
    _poolLimit = poolLimit;
    _samplePos = 0;

    _pDBMng = pDBMng;

    _modId = _pDBMng->getModId("CegoAdminThreadPool");

    pDBMng->getDBHost(_adminHostName);
    pDBMng->getAdminPort(_adminPortNo);

    _threadId = (long*)malloc(_poolLimit * sizeof(long));
    _threadLoad = (long*)malloc(_poolLimit * sizeof(long));
    _numRequest = (long*)malloc(_poolLimit * sizeof(long));
    for ( int i = 0; i < THRMNG_NUMLOADSAMPLE; i++ )
    {
	_threadIdle[i] = (long*)malloc(_poolLimit * sizeof(long));
    }
    _threadState = (ThreadState*)malloc(_poolLimit * sizeof(ThreadState));
    _threadList = (CegoAdminThread**)malloc(_poolLimit * sizeof(CegoAdminThread*));

    _terminated=false;

    long i=0;
    while ( i < _poolLimit )
    {
	
	_threadState[i] = READY;
	_threadList[i]= new CegoAdminThread(this, pDBMng, pDbPool, pLogPool);
	_threadId[i]=i;
	_numRequest[i]=0;
	_threadLoad[i]=0;
	for ( int j = 0; j < THRMNG_NUMLOADSAMPLE; j++ )
	{
	    _threadIdle[j][i]=0;
	}

	_threadList[i]->start(&_threadId[i]);
	
	i++;
    }
}

CegoAdminThreadPool::~CegoAdminThreadPool()  
{
    _terminated=true;
    _joined=false;
    int count=0;
    while ( _joined == false && count < POOL_TERMWAIT )
    {
	Sleeper s;
	s.secSleep(1);
	count++;
    }
    
    if ( _joined )
    {
	_pDBMng->log(_modId, Logger::NOTICE, Chain("All admin threads terminated"));
	join(getTid());
    }
    else
    {
	_pDBMng->log(_modId, Logger::NOTICE, Chain("Canceling hanging admin sessions ..."));
	cancel();
    }

    long i=0;
    while ( i < _poolLimit )
    {
	delete _threadList[i];
	i++;
    }

    delete _threadLoad;
    for ( int i = 0; i < THRMNG_NUMLOADSAMPLE; i++ )
    {
	delete _threadIdle[i];
    }


    delete _threadId;
    delete _numRequest;
    delete _threadState;

}

int CegoAdminThreadPool::getPoolLimit() const
{
    return _poolLimit;
}

void CegoAdminThreadPool::getThreadInfo(long i, long& numRequest, long& threadLoad, CegoAdminThreadPool::ThreadState& state, Chain& action)
{
    state = _threadState[i];
    numRequest = _numRequest[i];
    threadLoad = _threadLoad[i];
    action = _threadList[i]->lastAction();
}


void CegoAdminThreadPool::setState(long i, CegoAdminThreadPool::ThreadState state)
{
    _threadState[i] = state;
}

void CegoAdminThreadPool::incNumRequest(long i)
{
    _numRequest[i]++;
}

void CegoAdminThreadPool::addThreadIdle(long i, long usec)
{
    _threadIdle[_samplePos][i] += usec;
}


void CegoAdminThreadPool::setTid(long i, long tid)
{
    _threadId[i] = tid;
}

void* CegoAdminThreadPool::job(void* arg)
{

    try {


	NanoTimer tim;
    
	Net net( NETMNG_MSG_BUFLEN, NETMNG_SIZEBUFLEN );
	
#ifdef CGDEBUG
	_pDBMng->log(_modId, Logger::DEBUG, Chain("Serving admin requests on port ") + Chain(_adminPortNo));
#endif

	// cout << "Serving portno " << _adminPortNo << endl;
	net.serve(_adminHostName, _adminPortNo);
	
	long usedTime[THRMNG_NUMLOADSAMPLE];
	for ( int i = 0; i < THRMNG_NUMLOADSAMPLE; i++ )
	{
	    usedTime[i] = 0;
	}

	while ( ! _terminated ) 
	{	    
	    // take the mutex to avoid cpu load during wait position

	    usedTime[_samplePos] = 0;
	    
	    tim.reset();
	    tim.start();

	
	    bool noReq=false;
	    lockQueue();
	    if ( _requestQueue.Size() == 0  )  noReq=true;
	    unlockQueue();
	    
	    if ( noReq )
		lockQueue();
	    
	    NetHandler* pHandle;
	    
	    try {

		pHandle = net.nextRequest( NETMNG_SELECT_TIMEOUT );

	    }
	    catch ( Exception e)
	    {
		Chain msg;
		e.pop(msg);
		_pDBMng->log(_modId, Logger::LOGERR, Chain("Error on incoming connection : ") +  msg);
	    }
	
	    
	    if ( noReq )
	    {
		unlockQueue();
		
		// in the following, we go sleep for a while
		// to give other threads a chance to aquire queue lock
		// on solaris without this sleep, bad scheduling behaviour happened

		Sleeper s;
		s.nanoSleep(NETMNG_QUEUE_DELAY);
	    }


	    if ( pHandle )
	    {

		// pHandle->setSendTimeout(NETMNG_SEND_TIMEOUT);
		// pHandle->setRecvTimeout(NETMNG_RECV_TIMEOUT);

		lockQueue();
		_requestQueue.Insert(pHandle);
		unlockQueue();
	    }

	    tim.stop();
	    usedTime[_samplePos] += tim.getSum();

	    tim.reset();
	    tim.start();
	   
	    // calculate load
	    long i=0;
	    // cout << "___" << endl;
	    // cout << "UsedTime=" << usedTime << endl;
	    while ( i < _poolLimit )
	    {
		long ut=0;
		long ti=0;
		for ( int j=0; j< THRMNG_NUMLOADSAMPLE; j++ )
		{
		    ut+=usedTime[j];
		    ti+=_threadIdle[j][i];
		}
		 
		long l = 0;
		if ( ut > 0 )
		    l = 100 - ( 100 * ti ) / ut;
		_threadLoad[i] = l > 0 ? l : 0; 

		i++;
	    }

	    _samplePos = ( _samplePos + 1 ) % THRMNG_NUMLOADSAMPLE;
	    
	    for ( int i = 0; i < _poolLimit; i++ )
	    {
		_threadIdle[_samplePos][i]=0;
	    }

	}

	// to terminate active recovery session
	_pDBMng->setAllRecoveryOff();

	int i=0;
	while ( i < _poolLimit )
	{

#ifdef CGDEBUG
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Waiting for admin thread with tid ") 
			 + Chain(_threadList[i]->getTid()) + Chain(" to terminate"));
#endif


	    join(_threadList[i]->getTid());
	    i++;
	}
	_joined=true;
    }
    catch ( Exception e)
    {
	Chain msg;
	e.pop(msg);
	_pDBMng->log(_modId, Logger::LOGERR, Chain("Admin Thread start failed : ") +  msg);
	_terminated=true;
    }
    return 0;
}

bool CegoAdminThreadPool::isTerminated()
{
    return _terminated;
}

NetHandler* CegoAdminThreadPool::nextRequest()
{

    lockQueue();

    NetHandler **pRequest = _requestQueue.First();

    if ( pRequest )
    {

	NetHandler *pN = *pRequest;
	_requestQueue.Remove(*pRequest);
	unlockQueue();

	return pN;
    }
    else
    {
	unlockQueue();
	return 0;
    }    
}


void CegoAdminThreadPool::lockQueue()
{
    queueLock.writeLock();
}

void CegoAdminThreadPool::unlockQueue()
{
    queueLock.unlock();
}

void CegoAdminThreadPool::getLockStat(Chain& lockName, long& lockCount, long &numRdLock, long &numWrLock, long &sumRdDelay, long &sumWrDelay)
{
    lockName = queueLock.getId();
    lockCount = queueLock.numLockTry();

    numRdLock = queueLock.numReadLock();
    numWrLock = queueLock.numWriteLock();
    sumRdDelay = 0;
    sumWrDelay = 0;

    if ( queueLock.numReadLock() > 0 )
	sumRdDelay = queueLock.sumReadDelay() / LCKMNG_DELRES;
    if ( queueLock.numWriteLock() > 0 )
	sumWrDelay = queueLock.sumWriteDelay() / LCKMNG_DELRES;

}
