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

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

// xml includes
#include <lfc/XMLSuite.h>

// cego includes
#include "CegoDefs.h"
#include "CegoXMLdef.h"
#include "CegoLogThreadPool.h"
#include "CegoLogRecord.h"

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

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

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

CegoLogThreadPool::CegoLogThreadPool(CegoDatabaseManager *pDBMng) : Thread()
{
    queueLock.init(LCKMNG_LOCKWAITDELAY, __lockStatOn);

    _pDBMng = pDBMng;
    _poolLimit = 0;
    _samplePos = 0;    
    _terminated=false;

    _modId = _pDBMng->getModId("CegoLogThreadPool");
}

CegoLogThreadPool::CegoLogThreadPool(int poolLimit, CegoDatabaseManager *pDBMng) : Thread()
{
    queueLock.init(LCKMNG_LOCKWAITDELAY, __lockStatOn);

    _pDBMng = pDBMng;
    _poolLimit = poolLimit;
    _samplePos = 0;    

    pDBMng->getDBHost(_logHostName);
    pDBMng->getLogPort(_logPortNo);

    _threadId = (long*)malloc(_poolLimit * sizeof(long));

    _threadLoad = (long*)malloc(_poolLimit * sizeof(long));
    for ( int i = 0; i < THRMNG_NUMLOADSAMPLE; i++ )
    {
	_threadIdle[i] = (long*)malloc(_poolLimit * sizeof(long));
    }

    _numRequest = (long*)malloc(_poolLimit * sizeof(long));
    _threadState = (ThreadState*)malloc(_poolLimit * sizeof(ThreadState));
    _threadList = (CegoLogThread**)malloc(_poolLimit * sizeof(CegoLogThread*));

    _terminated=false;

    long i=0;
    while ( i < _poolLimit )
    {
	
	_threadState[i] = READY;
	_threadList[i]= new CegoLogThread(this, pDBMng);
	_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++;
    }

    _modId = _pDBMng->getModId("CegoLogThreadPool");
}

CegoLogThreadPool::~CegoLogThreadPool()  
{
    _joined=false;
    _terminated=true;

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

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

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

	delete _numRequest;
	delete _threadState;
    }
    
}

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

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

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

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

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


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

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

    // special mode for batch import 
    if ( _poolLimit == 0 )
    {
	while ( ! _terminated )
	{
	    try 
	    {
		shiftRedoLogs();
		Sleeper s;
		s.secSleep(1);
	    }
	    catch ( Exception e)
	    {
		Chain msg;
		e.pop(msg);
		_pDBMng->log(_modId, Logger::LOGERR, Chain("Log thread failed : ") + msg); 
		// _terminated=true;
	    }

	}
	_joined=true;	

    }
    else
    {
	try {

	    NanoTimer tim;
	    
	    Net net ( NETMNG_MSG_BUFLEN, NETMNG_SIZEBUFLEN );
	    
	    net.serve(_logHostName, _logPortNo);

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

		    Sleeper s;
		    s.nanoSleep(NETMNG_QUEUE_DELAY);

		}
		
		if ( pHandle )
		{
		    
		    // pHandle->setSendTimeout(NETMNG_SEND_TIMEOUT);
		    // pHandle->setRecvTimeout(NETMNG_RECV_TIMEOUT);
		    
		    lockQueue();
		    _requestQueue.Insert(pHandle);
		    unlockQueue();
		}
		
		try
		{
		    // next job of the main thread is to archive redologs	    
		    shiftRedoLogs();
		}
		catch ( Exception e)
		{
		    Chain msg;
		    e.pop(msg);
		    _pDBMng->log(_modId, Logger::LOGERR, Chain("Log thread failed : ") + msg); 
		    // _terminated=true;
		}

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

	    int i=0;
	    while ( i < _poolLimit )
	    {
		_pDBMng->log(_modId, Logger::DEBUG, Chain("Waiting for log tread 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("Log thread start failed : ") + msg); 
	    _terminated=true;
	}
    }
    return 0;
}

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

NetHandler* CegoLogThreadPool::nextRequest()
{
    lockQueue();
    NetHandler **pRequest = _requestQueue.First();
    if ( pRequest )
    {
	NetHandler *pN = *pRequest;
	_requestQueue.Remove(*pRequest);
	unlockQueue();
	return pN;
    }
    else
    {
	unlockQueue();
	return 0;
    }    
}

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

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

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

}

void CegoLogThreadPool::shiftRedoLogs()
{
    
    ListT<Chain> tsList;

    Host h;
    _pDBMng->getActiveTableSet(h.getName(), tsList); 

    Chain *pActiveTS = tsList.First();

    while ( pActiveTS )
    {

	int tabSetId = _pDBMng->getTabSetId(*pActiveTS);
	
	ListT<Chain> archPathList;
	ListT<Chain> archIdList;
	ListT<Chain> occupiedLogList;
	
	_pDBMng->getArchLogInfo(tabSetId, archIdList, archPathList);
	_pDBMng->getOccupiedLogList(tabSetId, occupiedLogList);
	
	Chain *pOccupiedLog = occupiedLogList.First();
	
	while ( pOccupiedLog )
	{
	    Chain *pArchPath = archPathList.First();
	    
	    while ( pArchPath )
	    {		    
		copyLog(*pActiveTS, *pOccupiedLog, *pArchPath);
		pArchPath = archPathList.Next();
	    }
	    
	    _pDBMng->setLogFileStatus(tabSetId, *pOccupiedLog, XML_FREE_VALUE);			
	    _pDBMng->doc2Xml();
	    
	    pOccupiedLog = occupiedLogList.Next();
	}
    
	pActiveTS = tsList.Next();
    }
}

void CegoLogThreadPool::copyLog(const Chain& tableSet, const Chain& logFileName, const Chain& archLogPath)
{

    File logFile(logFileName);
    
    logFile.open(File::READ);

    int offset;
    int c = logFile.readByte((char*)&offset, sizeof(int));

    int pos = sizeof(int);

    if ( pos < offset)
    {
	// logFile.seek(sizeof(int));
	
	int len;
	c = logFile.readByte((char*)&len, sizeof(int));	
	
	char* buf = new char[len];
	
	logFile.readByte(buf, len);
	
	CegoLogRecord logRec;
	logRec.decode(buf);    
	delete buf;
	logFile.close();

	
	Chain archLogFile = _pDBMng->getArchiveLogName(tableSet, logRec.getLSN());
	
	File archFile(archLogPath + Chain("/") + archLogFile);

	_pDBMng->log(_modId, Logger::NOTICE, Chain("Archiving logfile ") + logFile.getFileName() + Chain(" to ") + archFile.getFileName()); 

	archFile = logFile;
	
	archFile.open(File::APPEND);
	archFile.trunc(offset);
    }

    return;

}
