///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoDbThreadPool.cc  
// -------------------                                                     
// Cego db 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: CegoDbThreadPool
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

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

#include <lfc/XMLSuite.h>

#include "CegoXMLdef.h"
#include "CegoDbThreadPool.h"
#include "CegoDefs.h"

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

static ThreadLock queueLock("DBQUEUE"); 
static ThreadLock** thrLockArray; 
extern bool __lockStatOn;

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

CegoDbThreadPool::CegoDbThreadPool(int poolLimit, CegoDatabaseManager *pDBMng, CegoDbHandler::ProtocolType protType)  : Thread()
{
    _poolLimit = poolLimit;
    _samplePos = 0;
    _pDBMng = pDBMng;
    _protType = protType;
    _modId = _pDBMng->getModId("CegoDbThreadPool");

}

CegoDbThreadPool::~CegoDbThreadPool()  
{
    _terminated=true;

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

    _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 db threads terminated"));
	join(getTid());
    }
    else
    {
	_pDBMng->log(_modId, Logger::NOTICE, Chain("Canceling hanging db sessions ..."));
	cancel();
    }
    
    for ( int i = 0; i < _poolLimit; i++ )
    {
	delete _threadList[i];
	delete thrLockArray[i];
    }
    delete _numRequest;
    delete _numQueryRequest;
    delete _threadId;
    delete _threadLoad;
    for ( int i = 0; i < THRMNG_NUMLOADSAMPLE; i++ )
    {
	delete _threadIdle[i];
    }
    delete _threadState;
    delete thrLockArray;

}

void CegoDbThreadPool::P(int i)
{
    thrLockArray[i]->writeLock(DBM_LOCKTIMEOUT);
}

void CegoDbThreadPool::V(int i)
{
    thrLockArray[i]->unlock();
}


void  CegoDbThreadPool::syncToReady()
{

    queueLock.init(LCKMNG_LOCKWAITDELAY, __lockStatOn);
    thrLockArray = new ThreadLock*[_poolLimit];

    for (int i = 0; i<_poolLimit; i++)
    {
        thrLockArray[i] = new ThreadLock(Chain("THRLCK") + Chain(i));
        thrLockArray[i]->init(LCKMNG_LOCKWAITDELAY, __lockStatOn);
    }
    
    _pDBMng->getDataPort(_dataPortNo);
    _pDBMng->getDBHost(_dataHostName);

    _threadId = (long*)malloc(_poolLimit * sizeof(long));
    _threadLoad = (long*)malloc(_poolLimit * sizeof(long));
    _numRequest = (long*)malloc(_poolLimit * sizeof(long));
    _numQueryRequest = (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 = (CegoDbThread**)malloc(_poolLimit * sizeof(CegoDbThread*));

    _terminated=false;

    long i=0;
    while ( i < _poolLimit )
    {
	_threadState[i] = STARTING;
	_threadList[i]= new CegoDbThread(this, _pDBMng, _protType);
	_numRequest[i]=0;
	_numQueryRequest[i]=0;

	_threadId[i]=i;
	_threadLoad[i]=0;
	for ( int j = 0; j < THRMNG_NUMLOADSAMPLE; j++ )
	{
	    _threadIdle[j][i]=0;
	}
	_threadList[i]->start(&_threadId[i]);	
	i++;
    }
    
    int numReady=0;
    while ( numReady < _poolLimit )
    {	
	numReady=0;
	int i=0;
	while ( i < _poolLimit )
	{
	    if ( _threadState[i] == READY )
		numReady++;
	    i++;
	}
	_pDBMng->log(_modId, Logger::NOTICE, Chain(numReady) + Chain(" of " ) + Chain(_poolLimit) + Chain(" db threads ready"));
	
	Sleeper s;
	s.milliSleep(1);

    }
    _pDBMng->log(_modId, Logger::NOTICE, Chain("All db threads ready"));
}

void  CegoDbThreadPool::loadObjects(int tabSetId)
{
    int i=0;
    while ( i < _poolLimit )
    {
	_threadList[i]->addReloadRequest(tabSetId);
	i++;
    }
}

void  CegoDbThreadPool::unloadObjects(int tabSetId)
{
    int i=0;
    while ( i < _poolLimit )
    {
	_threadList[i]->unloadObjects(tabSetId);
	i++;
    }
}

void CegoDbThreadPool::invalidateObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    int i=0;
    while ( i < _poolLimit )
    {	
	try
	{
	    _threadList[i]->invalidateObject(tabSetId, objName, type);
	}
	catch ( Exception e )
	{
	    Chain msg;
	    _pDBMng->log(_modId, Logger::NOTICE, Chain("Ignoring error : ") + msg);
	}
	i++;
    }    
}

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

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

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

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

void CegoDbThreadPool::incNumQueryRequest(long i)
{
    _numQueryRequest[i]++;
}

void CegoDbThreadPool::addThreadIdle(long i, long usec)
{
    // Chain s = Chain("Thread ") + Chain(i) + Chain(" pos ") + Chain(_samplePos) + Chain(" to ") + Chain(usec);
    // cout << s << endl;

    _threadIdle[_samplePos][i] += usec;
}


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

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

void  CegoDbThreadPool::abortThread(long i)
{
    _threadList[i]->abortSession();
}

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

    try {

	NanoTimer tim;
	
	Net net( NETMNG_MSG_BUFLEN, NETMNG_SIZEBUFLEN );
	
	// cout << "Serving portno " << _portNo << endl;
	net.serve(_dataHostName, _dataPortNo);

	long usedTime[THRMNG_NUMLOADSAMPLE];
	for ( int i = 0; i < THRMNG_NUMLOADSAMPLE; i++ )
	{
	    usedTime[i] = 0;
	}
	
	while ( ! _terminated ) 
	{

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

	    // take the mutex to avoid cpu load during wait position
	
	    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();
		
		// struct timespec delay;	
		// delay.tv_sec = 0; 
		// delay.tv_nsec = NETMNG_QUEUE_DELAY;
		// nanosleep(&delay, NULL);

		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;
	    }
	    

	}
	long i=0;
	while ( i < _poolLimit )
	{
	    _pDBMng->log(_modId, Logger::DEBUG, Chain("Waiting for db thread with tid ") + Chain(_threadList[i]->getTid()) + Chain(" to terminate"));
	    join(_threadList[i]->getTid());
	    i++;
	}
	_joined=true;
    }
    catch ( Exception e)
    {
	Chain msg;
	e.pop(msg);	
	_pDBMng->log(_modId, Logger::LOGERR, Chain("Db Thread start failed : ") +  msg);	
	_terminated=true;
    }

    // waiting for db threads to terminate
    
    return 0;
}

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

NetHandler* CegoDbThreadPool::nextRequest()
{

    lockQueue();

    NetHandler **pRequest = _requestQueue.First();

    if ( pRequest )
    {

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

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

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

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


void CegoDbThreadPool::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;

}
