///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoBufferPool.cc
// -----------------
// Cego buffer pool implementation module
//
// Design and Implementation by Bjoern Lemke               
//     
// (C)opyright 2000-2013 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: CegoBufferPool
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

#include <lfc/Exception.h>
#include <lfc/CommandExecuter.h>
#include <lfc/Datetime.h>
#include <lfc/Sleeper.h>

#include "CegoBufferPool.h"
#include "CegoDefs.h"

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

#define BUFFERPOOLHEAD (((sizeof(BufferPoolHead)-1)/BUPMNG_ALIGNMENT)+1)*BUPMNG_ALIGNMENT
#define BUFFERHEAD (((sizeof(BufferHead)-1)/BUPMNG_ALIGNMENT)+1)*BUPMNG_ALIGNMENT


CegoBufferPool::CegoBufferPool(const Chain& xmlDef, const Chain& logFile) : CegoLogManager(xmlDef, logFile)
{    
    _pBufPool = 0;
    _numDiskRead=0;
    _numDiskWrite=0;
    _fixCount=0;
    _avgReadDelay=0;
    _avgWriteDelay=0;

    Datetime ts;
    _statStart = ts.asInt(); 
    _modId = getModId("CegoBufferPool");
}

CegoBufferPool::~CegoBufferPool()
{
}

void CegoBufferPool::initPool(long numPages)
{

#ifdef DEBUG
    log(_modId, Logger::DEBUG, Chain("Reading xml def ..."));
#endif

    xml2Doc();

    initDoc();
    
    Chain dbName = getDbName();
    int pageSize = getPageSize();
    _maxFixTries = getMaxFixTries();

    _dbName = dbName;
    _numPages = numPages;
    _pageSize = pageSize;

    if (_pBufPool == 0)
    {
	
	log(_modId, Logger::NOTICE, Chain("Allocating ") + Chain(_numPages) + Chain(" buffer pool pages ..."));

	_pBufPool = malloc(_numPages * (BUFFERHEAD + _pageSize) + BUFFERPOOLHEAD);

	if ( _pBufPool == NULL )
	{
	    throw Exception(EXLOC, "Cannot initialize pool");
	}

	log(_modId, Logger::NOTICE, Chain("Initializing buffer pool pages ..."));
	
	void *base = (void*)_pBufPool;
	void *ptr = (void*)_pBufPool;

	BufferPoolHead bph;
	bph.numPages=_numPages;
	
	memcpy(base, &bph, sizeof(BufferPoolHead));
	
	ptr = (void*)(BUFFERPOOLHEAD + (long)base);	

	BufferHead ibh;
	ibh.isOccupied = NOT_OCCUPIED;
	ibh.isDirty = 0;
	ibh.numFixes = 0;
	
	for (int i = 0; i<_numPages; i++)
	{
	    memcpy(ptr, &ibh,  sizeof(BufferHead) );
	    ptr = (void*)((long)ptr + BUFFERHEAD + _pageSize);
	}
	
	log(_modId, Logger::NOTICE, Chain("Pool started"));
    }
    else
    {
	throw Exception(EXLOC, "Buffer pool already created");
    }
    
}

void CegoBufferPool::removePool()
{
    if (_pBufPool)
    {
	log(_modId, Logger::NOTICE, Chain("Stopping pool ..."));
	free ( _pBufPool );
	_pBufPool = 0;
	log(_modId, Logger::NOTICE, Chain("Pool stopped"));
    }
}

Chain CegoBufferPool::getDBName()
{
    return _dbName;
}

void CegoBufferPool::bufferFix(CegoBufferPage &bp, int tabSetId, int fileId, int pageId, FixMode m, CegoLockHandler *pLockHandle, int numTry)
{

    _fixCount++;

    if ( _fixCount % BUPMNG_STATSPERIOD == 0 )
    {
	resetStats();
    }

    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }

    void *base = (void*)_pBufPool;

    int hashId = ( (fileId+1) * (pageId+1) ) % _numPages;

    void* bufPtr = (void*)((long)base + BUFFERPOOLHEAD + hashId * ( BUFFERHEAD + _pageSize ));
    
    bool isFixed = false;
    
    int numTries = 0;
    
    // Step I : searching for possible slot

    
    while ( ! isFixed && numTries < _maxFixTries )
    {

	pLockHandle->lockBufferPool(hashId, CegoLockHandler::WRITE);

	BufferHead bh;
	memcpy(&bh, bufPtr, sizeof(BufferHead));

	int minFixStat = -1;

	if ( bh.isOccupied != NOT_OCCUPIED && bh.pageId == pageId && bh.fileId == fileId)
	{
	    isFixed = true;

	    if ( m == CegoBufferPool::PERSISTENT )
		bh.isOccupied = PERSISTENT_OCCUPIED;
	    else if ( m == CegoBufferPool::NOSYNC && bh.isOccupied == WRITE_ON_SYNC)
		bh.isOccupied = WRITE_ON_DIRTY;
	    else
	    {
		bh.isOccupied = WRITE_ON_SYNC;
	    }
	    
	    bh.numFixes = bh.numFixes + 1;
	    bh.fixStat++;

	    memcpy(bufPtr, &bh, sizeof(BufferHead));
	    
	    pLockHandle->unlockBufferPool(hashId);

	}
	else 
	{

	    pLockHandle->unlockBufferPool(hashId);

	    hashId =  ( hashId + 1 ) % ( _numPages - 1 );
	    bufPtr = (void*)( hashId * ( BUFFERHEAD + _pageSize ) + (long)base + BUFFERPOOLHEAD );
	    numTries++;
	}
	
    }

    // Step II : page is not in buffercache yet, searching free and not occupied slot 

    void* minFixStatBufPtr = 0;
    int minHashId = -1;

    if ( ! isFixed )
    {
	numTries = 0;    
	
	hashId = ( (fileId+1 ) * (pageId+1) ) % _numPages;
	bufPtr = (void*)((long)base + BUFFERPOOLHEAD + hashId * ( BUFFERHEAD + _pageSize ));

    }
    
    while ( ! isFixed && numTries < _maxFixTries )
    {

	try
	{
	    pLockHandle->lockBufferPool(hashId, CegoLockHandler::WRITE);
	}
	catch ( Exception e )
	{
	    if ( minHashId != -1 )
		pLockHandle->unlockBufferPool(minHashId);		
	    throw Exception(EXLOC, "Cannot fix buffer", e);
	}
	
	// cout << "Trying to fix with try " << numTries << endl;
	BufferHead bh;
	memcpy(&bh, bufPtr, sizeof(BufferHead));
	
	int minFixStat = -1;
	
	if ( bh.isOccupied == NOT_OCCUPIED )
	{
	    if ( m == CegoBufferPool::PERSISTENT )
		bh.isOccupied = PERSISTENT_OCCUPIED;
	    else if ( m == CegoBufferPool::SYNC )
		bh.isOccupied = WRITE_ON_SYNC;
	    else 
		bh.isOccupied = WRITE_ON_DIRTY;

	    bh.pageId = pageId;
	    bh.fileId = fileId;
	    bh.isDirty = 0;
	    bh.numFixes = 1;
	    
	    try
	    {		
		
		_numDiskRead++;
		_diskReadTimer.start();
		readPage(fileId, pageId, bh.tabSetId, bh.fixStat, (char*)((long)bufPtr+BUFFERHEAD), pLockHandle);
		_diskReadTimer.stop();
		_avgReadDelay = ( _diskReadTimer.getSum() /  (_numDiskRead+(long)1)  ) / 1000;

	    }
	    catch ( Exception e )
	    {
		pLockHandle->unlockBufferPool(hashId);
		if ( minHashId != -1 )
		    pLockHandle->unlockBufferPool(minHashId);		
		throw Exception(EXLOC, "Cannot fix buffer", e);
	    }

	    bh.fixStat++;
	    memcpy(bufPtr, &bh, sizeof(BufferHead));	
	    
	    isFixed = true;

	    pLockHandle->unlockBufferPool(hashId);
	    if ( minHashId != -1 )
		pLockHandle->unlockBufferPool(minHashId);
	    
	}
	else 
	{

	    // for now, we take any tabSetId
	    // tabSetId = bh.tabSetId;
	    
	    if ( ( ( bh.isOccupied != PERSISTENT_OCCUPIED && bh.isDirty == 0 )
		   || bh.isOccupied == WRITE_ON_DIRTY )
		 && bh.numFixes == 0 && ( minFixStat > bh.fixStat || minFixStat == -1))
	    {
		minFixStatBufPtr = bufPtr;
		// cout << "Found page with minFixStat = " << bh.fixStat << endl;
		minFixStat = bh.fixStat;
		
		if ( minHashId != -1 )
		{
		    pLockHandle->unlockBufferPool(minHashId);	   
		}
		
		minHashId = hashId;

		try
		{
		    pLockHandle->lockBufferPool(minHashId, CegoLockHandler::WRITE);
		}
		catch ( Exception e )
		{
		    pLockHandle->unlockBufferPool(hashId);
		    throw Exception(EXLOC, "Cannot fix buffer", e);
		}
	    }
	    
	    pLockHandle->unlockBufferPool(hashId);
	    
	    hashId =  ( hashId + 1 ) % ( _numPages - 1 );
	    bufPtr = (void*)( hashId * ( BUFFERHEAD + _pageSize ) + (long)base + BUFFERPOOLHEAD );
	    numTries++;
	}
	
    }

    // Step III : page is not in buffercache yet and all slots are occupied, using the occupied slot
    //            with lowest fix stat 
    
    if ( ! isFixed && minFixStatBufPtr != 0)
    {
	
	bufPtr = minFixStatBufPtr;

	// minHashId is already locked
	// pLockHandle->lockBufferPool(minHashId);
		
	BufferHead bh;
	memcpy(&bh, bufPtr, sizeof(BufferHead));
	
	if ( bh.isOccupied == WRITE_ON_DIRTY &&  bh.isDirty == 1 )
	{
	    _numDiskWrite++;
	    
#ifdef DEBUG
	    log(_modId, Logger::DEBUG, Chain("Async write of page [") + Chain(bh.fileId) + Chain(",") + Chain(bh.pageId) + Chain("]"));
#endif

	    // cout << "---- >>>> Async write of page [" << bh.fileId << "," << bh.pageId << "]" << endl;

	    _diskWriteTimer.start();
	    writePage(bh.fileId, bh.pageId, bh.fixStat, (char*)((long)bufPtr + BUFFERHEAD), pLockHandle);
	    _diskWriteTimer.stop();

	    _avgWriteDelay = ( _diskWriteTimer.getSum() /  (_numDiskWrite+(long)1)  ) / 1000;

	}
	
	// cout << "Using occupied page with fileId " << bh.fileId << " pageId " << bh.pageId << " numFixes " << bh.numFixes << endl;

	if ( m == CegoBufferPool::PERSISTENT )
	    bh.isOccupied = PERSISTENT_OCCUPIED;
	else if (m == CegoBufferPool::SYNC )
	    bh.isOccupied = WRITE_ON_SYNC;
	else if (m == CegoBufferPool::NOSYNC )
	    bh.isOccupied = WRITE_ON_DIRTY;

	bh.pageId = pageId;
	bh.fileId = fileId;
	bh.isDirty = 0;
	bh.numFixes = 1;
	
	try
	{
	    _numDiskRead++;
	    _diskReadTimer.start();
	    readPage(fileId, pageId, bh.tabSetId, bh.fixStat, (char*)((long)bufPtr+BUFFERHEAD), pLockHandle);
	    _diskReadTimer.stop();

	    _avgReadDelay = ( _diskReadTimer.getSum() /  (_numDiskRead+(long)1) ) / 1000;

	}
	catch ( Exception e )
	{
	    pLockHandle->unlockBufferPool(minHashId);
	    throw Exception(EXLOC, "Cannot fix buffer", e);
	}
	bh.fixStat++;
	memcpy(bufPtr, &bh, sizeof(BufferHead));
	
	pLockHandle->unlockBufferPool(minHashId);

	isFixed = true;
	
    }

    if ( ! isFixed )
    {

	log(_modId, Logger::NOTICE, Chain("Forced checkpoint for tabSetId ") + Chain(tabSetId) + Chain(" at lsn=") + Chain(getLSN(tabSetId)));

	writeCheckPoint(tabSetId, true, Chain(""), 0,  pLockHandle);

	// try again
	
	// if ( numTry > 5 )
	//     throw Exception(EXLOC, "Cannot fix buffer");

	numTry++;
	return bufferFix(bp, tabSetId, fileId, pageId, m, pLockHandle, numTry);
	
    }

    bp.setPageSize(_pageSize);
    bp.setPagePtr((char*)((long)bufPtr+BUFFERHEAD));
    bp.setPageHead((CegoBufferPage::PageHead*)((long)bufPtr+BUFFERHEAD));
    
    // CegoBufferPage* pBP = new CegoBufferPage((void*)((long)bufPtr+BUFFERHEAD), _pageSize);
    bp.setFileId(fileId);
    bp.setPageId(pageId);
    bp.setFixed(true);

}


void CegoBufferPool::emptyFix(CegoBufferPage &bp, int tabSetId, FixMode m, CegoFileHandler::FileType ft, CegoLockHandler *pLockHandle, bool doAppend)
{


    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }
    
    int fileId, pageId;
    
    unsigned* fbm;
    int fbmSize=0;
    
    allocatePage(tabSetId, ft, fileId, pageId, pLockHandle, fbm, fbmSize, doAppend);
    
    if ( fbmSize > 0 )
    {
	logBM(tabSetId, fileId, fbm, fbmSize);
	delete fbm;
    }
    
    bufferFix(bp, tabSetId, fileId, pageId, m, pLockHandle);
    bp.initPage(CegoBufferPage::TABLE);
    bp.setFixed(true);
}

void CegoBufferPool::bufferUnfix(CegoBufferPage &bp, bool isDirty, CegoLockHandler *pLockHandle)
{

    // cout << "BufferUnfix of page " << bp.getFileId() << "," << bp.getPageId() << endl;
    

    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }
    
    void *base = (void*)_pBufPool;
    char* bufPtr = (char*)bp.getPagePtr();
    int hashId = ( (long)bufPtr - (long)base - BUFFERPOOLHEAD ) / (  BUFFERHEAD + _pageSize );
    
    bufPtr = bufPtr - BUFFERHEAD;
    
    BufferHead bh;
    
    pLockHandle->lockBufferPool(hashId, CegoLockHandler::WRITE);
    
    memcpy(&bh, bufPtr, sizeof(BufferHead));
    
    // cout << "NumFixes = " << bh.numFixes << endl;
    
    if (bh.numFixes > 0 )
    {
	if (isDirty)
	{
	    bh.isDirty = 1;
	}
	bh.numFixes--;
    }
    else
    {	
	pLockHandle->unlockBufferPool(hashId);
	throw Exception(EXLOC, "Number of fixes is already zero");
    }
    memcpy(bufPtr, &bh, sizeof(BufferHead));
    
    bp.setFixed(false);
    
    pLockHandle->unlockBufferPool(hashId);
}    


void CegoBufferPool::bufferRelease(CegoBufferPage &bp, CegoLockHandler *pLockHandle)
{

    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }
    
    char* bufPtr = (char*)bp.getPagePtr();
    bufPtr = bufPtr - BUFFERHEAD;
    
    BufferHead bh;
    
    int hashId = ( (bp.getFileId()+1) * (bp.getPageId()+1) ) % _numPages;
    
    pLockHandle->lockBufferPool(hashId, CegoLockHandler::WRITE);
    
    try
    {
	memcpy(&bh, bufPtr, sizeof(BufferHead));
	
	bh.isOccupied = NOT_OCCUPIED;
	bh.numFixes=0;
	
	memcpy(bufPtr, &bh, sizeof(BufferHead));
	
	int fbmSize=0;
	unsigned* fbm;
	
	releasePage(bp.getFileId(), bp.getPageId(), pLockHandle, fbm, fbmSize);
	
	if ( fbmSize > 0 )
	{
	    logBM(bh.tabSetId, bp.getFileId(), fbm, fbmSize);
	    delete fbm;
	}
	
	pLockHandle->unlockBufferPool(hashId);
	
    }
    catch ( Exception e )
    {
	pLockHandle->unlockBufferPool(hashId);
	throw Exception(EXLOC, "Cannot release file page", e);
    }
    bp.setFixed(false);    
}

int CegoBufferPool::writeCheckPoint(int tabSetId, bool switchLog, const Chain& escCmd, int timeout, CegoLockHandler *pLockHandle)
{
    
    pLockHandle->lockBufferPool();
    
    int lsn;
    try
    {
	
	if (_pBufPool == 0)
	{
	    throw Exception(EXLOC, "No valid bufferpool" );
	}
	
	
	if ( escCmd != Chain("") )
	{
	    log(_modId, Logger::NOTICE, Chain("Executing escape command <") + escCmd + Chain(">"));
	    

	    char *pShell = getenv(CGEXESHELLVARNAME);
	    if ( pShell == NULL )
	    {
		throw Exception(EXLOC, "Execution shell not set");  		    
	    }

	    CommandExecuter cmdExe(pShell);
	    
	    try 
	    {
		int retCode = cmdExe.execute(escCmd, timeout);
		log(_modId, Logger::NOTICE, Chain("Escape command finished with return code : <") + Chain(retCode) + Chain(">"));
	    }
	    catch ( Exception e )
	    {
		Chain msg;
		e.pop(msg);
		log(_modId, Logger::LOGERR, msg);
		throw Exception(EXLOC, msg);
	    }
	}
	
	if ( switchLog )
	{
	    while ( switchLogFile(tabSetId) == false )
	    {
		
		log(_modId, Logger::NOTICE, Chain("Logfile for tabSetId ") + Chain(tabSetId) + Chain(" still active, switch failed"));
		
		Sleeper s;
		s.secSleep(LOGMNG_LOGSWITCH_WAIT_DELAY);
		
	    }
	}
	
	lsn = getLSN(tabSetId);    
	
	if ( lsn > 0 )
	    lsn--;

	setCommittedLSN(tabSetId, lsn);
	
	doc2Xml();


	void *bufPtr = (void*)( (long)_pBufPool + BUFFERPOOLHEAD );
	
	int pageCount = 0;
	
	while ( pageCount < _numPages )
	{
	    BufferHead bh;
	    memcpy(&bh, bufPtr, sizeof(BufferHead));
	    
	    if (bh.isDirty != 0 && bh.tabSetId == tabSetId )
	    {
		
		if ( isBackup(bh.fileId) )
		{
		    if (  isMarked(bh.fileId, bh.pageId) == false )
		    {
			
#ifdef DEBUG
			log(_modId, Logger::DEBUG, Chain("Reading page (") + Chain(bh.fileId) + Chain(",") + Chain(bh.pageId) + Chain(") to log ..."));
#endif	
			int ts;
			int fixStat;
			char *pageData = new char[_pageSize];
			readPage(bh.fileId, bh.pageId, ts, fixStat, pageData, pLockHandle);
			
			CegoLogRecord lr;
			lr.setAction(CegoLogRecord::LOGREC_BUPAGE);
			lr.setData(pageData);
			lr.setDataLen(_pageSize);
			lr.setFileId(bh.fileId);
			lr.setPageId(bh.pageId);
			
			if (  logAction(tabSetId, lr) == CegoLogManager::LOG_FULL )
			{
			    
			    int gid;
			    Chain logFile;
			    int logSize = 0 ;
			    
#ifdef DEBUG
			    log(_modId, Logger::DEBUG, Chain("Switching logFiles ..."));
#endif	    
			    
			    while ( switchLogFile(tabSetId) == false )
			    {
				log(_modId, Logger::NOTICE, Chain("Logfile for tabSetId ") + Chain(tabSetId) + Chain(" still active, switch failed"));
				
				Sleeper s;
				s.secSleep(LOGMNG_LOGSWITCH_WAIT_DELAY);
				
			    }
			    
			    if ( logAction(tabSetId, lr) != CegoLogManager::LOG_SUCCESS )
			    {
				throw Exception(EXLOC, "Cannot write to log");
			    }	    		
			}
		    }
		}
		
		_numDiskWrite++;
		_diskWriteTimer.start();
		writePage(bh.fileId, bh.pageId, bh.fixStat, (char*)((long)bufPtr + BUFFERHEAD), pLockHandle);
		_diskWriteTimer.stop();
		
		_avgWriteDelay = ( _diskWriteTimer.getSum() /  (_numDiskWrite+(long)1) ) / 1000;
		
		
		bh.isDirty = 0;
		memcpy(bufPtr, &bh, sizeof(BufferHead));	    
	    }
	    bufPtr = (void*)((long)bufPtr + (long)(BUFFERHEAD + _pageSize));
	    pageCount++;
	}
    }
    catch ( Exception e )
    {
	pLockHandle->unlockBufferPool();
	throw e;
    }
    
    pLockHandle->unlockBufferPool();    
    return lsn;
}

void CegoBufferPool::writeAndRemoveTabSet(int tabSetId, CegoLockHandler *pLockHandle)
{

    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }
    
    void *bufPtr = (void*)( (long)_pBufPool + BUFFERPOOLHEAD );
    
    int pageCount = 0;
    
    while ( pageCount < _numPages )
    {
	BufferHead bh;
	memcpy(&bh, bufPtr, sizeof(BufferHead));
	
	if ( bh.tabSetId == tabSetId )
	{
	    
	    
	    if ( bh.isDirty != 0 )
	    {
		_numDiskWrite++;
		
		_diskWriteTimer.start();
		writePage(bh.fileId, bh.pageId, bh.fixStat, (char*)((long)bufPtr + BUFFERHEAD), pLockHandle);
		_diskWriteTimer.stop();
		
		_avgWriteDelay = ( _diskWriteTimer.getSum() /  (_numDiskWrite+(long)1) ) / 1000;
	    }
	    
	    bh.isOccupied = NOT_OCCUPIED;
	    bh.numFixes=0;
	    
	    memcpy(bufPtr, &bh, sizeof(BufferHead));
	    
	}
	bufPtr = (void*)((long)bufPtr + (long)(BUFFERHEAD + _pageSize));
	pageCount++;
    }
    
    releaseFiles(tabSetId);
}

void CegoBufferPool::poolInfo(int& pageSize, 
			      long& numTotal, 
			      long& numUsed, 
			      long& numFree, 
			      long& numDirty, 
			      long& numFixes, 
			      long& numPersistent, 
			      long& numNoSync, 
			      long& numDiskRead, 
			      long& numDiskWrite,
			      long& hitRate,
			      long& readDelay,
			      long& writeDelay,
			      long& curFixCount,
			      long& maxFixCount,
			      int& statStart)
{

    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }
    
    pageSize = _pageSize;
    numTotal = _numPages;
    numUsed=0;
    numFree=0;
    numFixes=0;
    numDirty=0;
    numPersistent=0;
    numNoSync=0;
    
    void* bufPtr = (void*)( (long)_pBufPool + BUFFERPOOLHEAD);
    
    long pageCount = 0;
    
    while ( pageCount < _numPages )
    {    
	BufferHead bh;
	memcpy(&bh, bufPtr, sizeof(BufferHead));
	
	numFixes += (long)bh.numFixes;
	numDirty += (long)bh.isDirty;
	
	if (bh.isOccupied == NOT_OCCUPIED)
	    numFree++;
	else if (bh.isOccupied == WRITE_ON_SYNC)
	{
	    numUsed++;
	}
	else if (bh.isOccupied == WRITE_ON_DIRTY)
	{
	    numUsed++;
	    numNoSync++;
	}
	else if (bh.isOccupied == PERSISTENT_OCCUPIED)
	{
	    numUsed++;
	    numPersistent++;
	}   
	
	bufPtr = (void*)((long)bufPtr + (long)(BUFFERHEAD + _pageSize));     
	pageCount++;
    }
    
    numDiskWrite = _numDiskWrite;
    numDiskRead = _numDiskRead;
    
    hitRate = 100000 - _numDiskRead / ( _fixCount / 100000 + 1) ;
    
    readDelay = _avgReadDelay;
    writeDelay = _avgWriteDelay;
    statStart = _statStart;
    curFixCount = _fixCount;
    maxFixCount = BUPMNG_STATSPERIOD;
}


void CegoBufferPool::getPoolEntryList(ListT<CegoBufferPoolEntry>& entryList)
{

    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }
    
    void* bufPtr = (void*)( (long)_pBufPool + BUFFERPOOLHEAD);
    
    long pageCount = 0;
    
    entryList.Empty();
    
    while ( pageCount < _numPages )
    {    
	BufferHead bh;
	memcpy(&bh, bufPtr, sizeof(BufferHead));
	
	bool isDirty = false;
	if ( bh.isDirty )
	    isDirty=true;
	
	Chain occState;
	if (bh.isOccupied == NOT_OCCUPIED)
	    occState=Chain("NOTOCCUPIED");
	else if (bh.isOccupied == WRITE_ON_SYNC)
	    occState=Chain("WRITEONSYNC");
	else if (bh.isOccupied == WRITE_ON_DIRTY)
	    occState=Chain("WRITEONDIRTY");
	else if (bh.isOccupied == PERSISTENT_OCCUPIED)
	    occState=Chain("PERSISTENT");
	
	CegoBufferPoolEntry bpe(pageCount, occState, isDirty, bh.numFixes, bh.tabSetId, bh.fileId, bh.pageId, bh.fixStat);
	entryList.Insert(bpe);
	
	bufPtr = (void*)((long)bufPtr + (long)(BUFFERHEAD + _pageSize));     
	pageCount++;
    }
    
}

void CegoBufferPool::resetStats()
{
    _numDiskRead=0;
    _numDiskWrite=0;
    _fixCount=0;
    _avgReadDelay=0;
    _avgWriteDelay=0;
    _diskReadTimer.reset();
    _diskWriteTimer.reset();
    Datetime ts;
    _statStart = ts.asInt(); 
}


void CegoBufferPool::logBM(int tabSetId, int fileId, unsigned* fbm, int fbmSize)
{
    
    
    CegoLogRecord lr;
    
    lr.setAction(CegoLogRecord::LOGREC_BUFBM);
    lr.setData((char*)fbm);
    lr.setDataLen(fbmSize * sizeof(int));
    lr.setFileId(fileId);
    
    if ( logAction(tabSetId, lr) == CegoLogManager::LOG_FULL )
    {
	
	int gid;
	Chain logFile;
	int logSize = 0 ;
	
	while ( switchLogFile(tabSetId) == false )
	{
	    log(_modId, Logger::NOTICE, Chain("Logfile for tabSetId ") + Chain(tabSetId) + Chain(" still active, switch failed"));
	    Sleeper s;
	    s.secSleep(LOGMNG_LOGSWITCH_WAIT_DELAY);
	    
	}
	
	if ( logAction(tabSetId, lr) != CegoLogManager::LOG_SUCCESS )
	{
	    throw Exception(EXLOC, "Cannot write to log");
	}	    		
    }
}

void CegoBufferPool::printPool()
{

    if (_pBufPool == 0)
    {
	throw Exception(EXLOC, "No valid bufferpool" );
    }
    
    cout << "--- BufferPool ---" << endl;
    cout << "BasePtr: " << (long)_pBufPool << endl;
    cout << "PageSize: " << _pageSize << endl;
    cout << "NumPages: " << _numPages << endl;
    
    long numUsed=0;
    long numFree=0;
    long numFixes=0;
    long numDirty=0;

    void* bufPtr = (void*)( (long)_pBufPool + BUFFERPOOLHEAD);

    long pageCount = 0;
    
    while ( pageCount < _numPages )
    {    
	BufferHead bh;
	memcpy(&bh, bufPtr, sizeof(BufferHead));

	numFixes += (long)bh.numFixes;
	numDirty += (long)bh.isDirty;

	if (bh.isOccupied == 0)
	    numFree++;
	else
	    numUsed++;
	
	bufPtr = (void*)((long)bufPtr + (long)(BUFFERHEAD + _pageSize));     
	pageCount++;
    }
    
    cout << "NumUsed: " << numUsed << endl;
    cout << "NumFixes: " << numFixes << endl;
    cout << "NumDirty: " << numDirty << endl;
    cout << "NumFree: " << numFree << endl;

    cout << "NumFree: " << numFree << endl;

}







