///////////////////////////////////////////////////////////////////////////////
//                                                         
// CegoLockHandler.cc
// ------------------
// Cego database lock handler 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: CegoLockHandler
// 
// Description: 
//
// Status: QG-2.6
//
///////////////////////////////////////////////////////////////////////////////

#ifndef _REENTRANT
#define _REENTRANT    /* basic 3-lines for threads */
#endif

// base includes
#include <lfc/Exception.h>
#include <lfc/ThreadLock.h>
#include <lfc/Tokenizer.h>

// cego includes
#include "CegoLockHandler.h"
#include "CegoDatabaseManager.h"

#include <string.h>
#include <stdlib.h>
		 
static ThreadLock** _lockArray;

bool __lockStatOn = true;

CegoLockHandler::CegoLockHandler(CegoDatabaseManager* pDBMng)
{
    
    _numRecSema = pDBMng->getNumRecordSema();
    _numRBRecSema = pDBMng->getNumRBRecordSema();
    _numSysRecSema = pDBMng->getNumSysRecordSema();

    _numSysPageSema = pDBMng->getNumSysPageSema();
    _numDataPageSema = pDBMng->getNumDataPageSema();
    _numIdxPageSema = pDBMng->getNumIndexPageSema();
    _numRBPageSema = pDBMng->getNumRBPageSema();
    _numDataFileSema = pDBMng->getNumDataFileSema();
    _numBufferPoolSema = pDBMng->getNumBufferPoolSema();

    _pageLockTimeout = pDBMng->getPageLockTimeout();
    _recLockTimeout = pDBMng->getRecLockTimeout();
    _fileLockTimeout = pDBMng->getFileLockTimeout();
    _poolLockTimeout = pDBMng->getPoolLockTimeout();
    _numLockTries = pDBMng->getNumLockTries();

    _pDBMng = pDBMng;
    _modId = pDBMng->getModId("CegoLockHandler");

    _numSema = _numRecSema
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema
	+ _numDataPageSema
	+ _numIdxPageSema
	+ _numRBPageSema
	+ _numDataFileSema
	+ _numBufferPoolSema;
    
    _lockCount = new int[_numSema];

   
    _recordLock = (RecordLock*)malloc( LCKMNG_NUMLOCKS * sizeof( struct RecordLock));
    _rbRecordLock = (RecordLock*)malloc( LCKMNG_NUMLOCKS * sizeof( struct RecordLock));
    _sysRecordLock = (RecordLock*)malloc( LCKMNG_NUMLOCKS * sizeof( struct RecordLock));

    _sysPageLock = (PageLock*)malloc( LCKMNG_NUMLOCKS * sizeof( struct PageLock));
    _dataPageLock = (PageLock*)malloc( LCKMNG_NUMLOCKS * sizeof( struct PageLock));
    _idxPageLock = (PageLock*)malloc( LCKMNG_NUMLOCKS * sizeof( struct PageLock));
    _rbPageLock = (PageLock*)malloc( LCKMNG_NUMLOCKS * sizeof( struct PageLock));
    

    int i;
    for (int i = 0; i<_numSema; i++)
    {
	_lockCount[i]=0;
    }
    for (i = 0; i<LCKMNG_NUMLOCKS; i++)
    {
	_recordLock[i].semId=0;
	_recordLock[i].lockId=0;

	_rbRecordLock[i].semId=0;
	_rbRecordLock[i].lockId=0;

	_sysRecordLock[i].semId=0;
	_sysRecordLock[i].lockId=0;

	_sysPageLock[i].lockId=0;
	_sysPageLock[i].fileId=0;
	_sysPageLock[i].pageId=0;
	_sysPageLock[i].semId=0;

	_dataPageLock[i].lockId=0;
	_dataPageLock[i].fileId=0;
	_dataPageLock[i].pageId=0;
	_dataPageLock[i].semId=0;

	_idxPageLock[i].lockId=0;
	_idxPageLock[i].fileId=0;
	_idxPageLock[i].pageId=0;
	_idxPageLock[i].semId=0;

	_rbPageLock[i].lockId=0;
	_rbPageLock[i].fileId=0;
	_rbPageLock[i].pageId=0;
	_rbPageLock[i].semId=0;
		
    }
    _nextLockId = 1;

}

CegoLockHandler::~CegoLockHandler()
{
    delete _lockCount;
    delete _recordLock;
    delete _rbRecordLock;
    delete _sysRecordLock;

    delete _sysPageLock;
    delete _dataPageLock;
    delete _idxPageLock;
    delete _rbPageLock;
}

void CegoLockHandler::initLocks()
{

    _lockArray = new ThreadLock*[_numSema];

    for (int i = 0; i<_numSema; i++)
    {
	_lockArray[i] = new ThreadLock();
	_lockArray[i]->init(LCKMNG_LOCKWAITDELAY, __lockStatOn);
    }

    int low = 0;
    int high = _numRecSema;
    int semId;

    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(RECLOCKGROUP) +  Chain("-") + Chain(semId-low));
    }

    low = _numRecSema;
    high = _numRecSema 
	+ _numRBRecSema;

    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(RBRECLOCKGROUP) + Chain("-") + Chain(semId-low));
    }

    low = _numRecSema
	+ _numRBRecSema;
    high = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema;

    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(SYSRECLOCKGROUP) + Chain("-") + Chain(semId-low));
    }

    low = _numRecSema
	+ _numRBRecSema
	+ _numSysRecSema;

    high = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema;


    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(SYSPAGELOCKGROUP) + Chain("-") + Chain(semId-low));
    }
	
    low = _numRecSema
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema;
    high = _numRecSema
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema;

    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(DATAPAGELOCKGROUP) + Chain("-") + Chain(semId-low));
    }

    low = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema;
    high = _numRecSema
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema ;

    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(IDXPAGELOCKGROUP) + Chain("-") + Chain(semId-low));
    }

    low = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema
	+ _numIdxPageSema;
    high = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema
	+ _numRBPageSema;
    
    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(RBPAGELOCKGROUP) + Chain("-") + Chain(semId-low));
    }

    low = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema
	+ _numRBPageSema;
    high = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema
	+ _numRBPageSema 
	+ _numDataFileSema;

    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(DATAFILELOCKGROUP) + Chain("-") + Chain(semId-low));
    }
  

    low = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema
	+ _numRBPageSema 
	+ _numDataFileSema;

    high = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema
	+ _numIdxPageSema
	+ _numRBPageSema  
	+ _numDataFileSema 
	+ _numBufferPoolSema;

    for (  semId = low ; semId < high ; semId++ )
    {
	_lockArray[semId]->setId( Chain(POOLLOCKGROUP) + Chain("-") + Chain(semId-low));
    }
    
}

void CegoLockHandler::deleteLocks()
{
    if ( _lockArray != 0 )
    {
	for (int i = 0; i<_numSema; i++)
	    delete _lockArray[i];
	delete _lockArray;
	_lockArray = 0;
    }
}

CegoBufferPage CegoLockHandler::getPage(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _recordLock[i].lockId != lockId )
    {
	i++;
    }
    
    if (_recordLock[i].lockId == lockId)
    {
	return _recordLock[i].bp;
    }
    else
    {
	Chain msg = Chain("Invalid page lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

unsigned long CegoLockHandler::lockRecord(const CegoBufferPage &bp, const CegoDataPointer& dp, enum LockMode lockMode)
{

    int semId = ( dp.getFileId() * dp.getPageId() + dp.getOffset()) % _numRecSema;
    
    int timeout = _recLockTimeout;
    int numTries = _numLockTries;

    int i=0;
    while ( _recordLock[i].lockId != 0 && i < LCKMNG_NUMLOCKS)
    {
	i++;
    }
    if (i == LCKMNG_NUMLOCKS)
    {
	throw Exception(EXLOC, "Number of record locks exceeded");
    }
    _recordLock[i].lockId=_nextLockId;    
    _nextLockId++;

    _recordLock[i].semId=semId;
    if ( _lockCount[semId] == 0 )
    {
	while ( numTries > 0 )
	{
	    try
	    {		
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else
		    _lockArray[semId]->writeLock(timeout);
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;									
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock record failed for fileId ") 
			+ Chain(dp.getFileId()) 
			+ Chain(" pageId ") + Chain(dp.getPageId()) + Chain(" offset=") + Chain(dp.getOffset());

		    throw Exception(EXLOC, msg, e); 
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Lock record failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}

	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }
    
    _recordLock[i].dp=dp;
    _recordLock[i].bp=bp;

    return _recordLock[i].lockId;
    
}

void CegoLockHandler::unlockRecord(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _recordLock[i].lockId != lockId )
    {
	i++;
    }
    if (_recordLock[i].lockId == lockId)
    {
	
	int semId = _recordLock[i].semId;
	
	if ( _lockCount[semId] > 1 )
	{
	    _lockCount[semId]--;
	}
	else
	{
	    _lockArray[semId]->unlock();
	    _lockCount[semId]=0;
	}
	_recordLock[i].lockId = 0;
	_recordLock[i].semId = 0;
    }
    else
    {
	Chain msg = Chain("Invalid record lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

unsigned long CegoLockHandler::lockRBRecord(const CegoBufferPage &bp, const CegoDataPointer& dp, enum LockMode lockMode)
{

    int semId = _numRecSema
	+ ( dp.getFileId() * dp.getPageId() + dp.getOffset()) % _numRBRecSema;
    
    int timeout = _recLockTimeout;
    int numTries = _numLockTries;

    int i=0;
    while ( _rbRecordLock[i].lockId != 0 && i < LCKMNG_NUMLOCKS)
    {
	i++;
    }
    if (i == LCKMNG_NUMLOCKS)
    {
	throw Exception(EXLOC, "Number of rb record locks exceeded");    
    }
    _rbRecordLock[i].lockId=_nextLockId;
    _nextLockId++;


    _rbRecordLock[i].semId=semId;
    if ( _lockCount[semId] == 0 )
    {
	while ( numTries > 0 )
	{
	    try
	    {		
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else
		    _lockArray[semId]->writeLock(timeout);
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;									
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock rb record failed for fileId ") 
			+ Chain(dp.getFileId()) 
			+ Chain(" pageId ") + Chain(dp.getPageId()) + Chain(" offset=") + Chain(dp.getOffset());

		    throw Exception(EXLOC, msg, e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("rb lock record failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}

	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }
    
    _rbRecordLock[i].dp=dp;
    _rbRecordLock[i].bp=bp;

    
    return _rbRecordLock[i].lockId;
    
}

void CegoLockHandler::unlockRBRecord(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _rbRecordLock[i].lockId != lockId )
    {
	i++;
    }
    if (_rbRecordLock[i].lockId == lockId)
    {
	
	int semId = _rbRecordLock[i].semId;
	
	if ( _lockCount[semId] > 1 )
	{
	    _lockCount[semId]--;
	}
	else
	{
	    _lockArray[semId]->unlock();
	    _lockCount[semId]=0;
	}
	_rbRecordLock[i].lockId = 0;
	_rbRecordLock[i].semId = 0;
    }
    else
    {
	Chain msg = Chain("Invalid rb record lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

unsigned long CegoLockHandler::lockSysRecord(const CegoBufferPage &bp, const CegoDataPointer& dp, enum LockMode lockMode)
{

    int timeout = _recLockTimeout;
    int numTries = _numLockTries;

    int semId = _numRecSema
	+ _numRBRecSema
	+ ( dp.getFileId() * dp.getPageId() + dp.getOffset()) % _numSysRecSema;

    int i=0;
    while ( _sysRecordLock[i].lockId != 0 && i < LCKMNG_NUMLOCKS)
    {
	i++;
    }
    if (i == LCKMNG_NUMLOCKS)
    {
	throw Exception(EXLOC, "Number of sys record locks exceeded");    
    }
    _sysRecordLock[i].lockId=_nextLockId;
    _nextLockId++;

    _sysRecordLock[i].semId=semId;
    if ( _lockCount[semId] == 0 )
    {
	while ( numTries > 0 )
	{
	    try
	    {		
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else
		    _lockArray[semId]->writeLock(timeout);
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;									
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock sys record failed for fileId ") 
			+ Chain(dp.getFileId()) 
			+ Chain(" pageId ") + Chain(dp.getPageId()) + Chain(" offset=") + Chain(dp.getOffset());
		    throw Exception(EXLOC, msg, e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("SYS lock record failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}

	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }
    
    _sysRecordLock[i].dp=dp;
    _sysRecordLock[i].bp=bp;
    
    return _sysRecordLock[i].lockId;
    
}

void CegoLockHandler::unlockSysRecord(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _sysRecordLock[i].lockId != lockId )
    {
	i++;
    }
    if (_sysRecordLock[i].lockId == lockId)
    {
	
	int semId = _sysRecordLock[i].semId;
	
	if ( _lockCount[semId] > 1 )
	{
	    _lockCount[semId]--;
	}
	else
	{
	    _lockArray[semId]->unlock();
	    _lockCount[semId]=0;
	}
	_sysRecordLock[i].lockId = 0;
	_sysRecordLock[i].semId = 0;
    }
    else
    {
	Chain msg = Chain("Invalid sys record lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

unsigned long CegoLockHandler::lockSysPage(int fileId, int pageId, enum LockMode lockMode)
{

    int semId = _numRecSema 
	+ _numRBRecSema 
	+ _numSysRecSema 
	+ ( fileId * ( pageId + 1 ) )  % _numSysPageSema;

    int timeout = _pageLockTimeout;
    int numTries = _numLockTries;

    int i=0;
    while ( _sysPageLock[i].lockId != 0 && i < LCKMNG_NUMLOCKS)
    {
	i++;
    }
    if (i == LCKMNG_NUMLOCKS)
    {
	throw Exception(EXLOC, "Number of sys page locks exceeded");    
    }
    _sysPageLock[i].lockId = _nextLockId;    
    _nextLockId++;

    if ( _lockCount[semId] == 0 )
    {

	while ( numTries > 0 )
	{
	    try
	    {
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else 
		    _lockArray[semId]->writeLock(timeout);
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;									
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock sys page failed for fileId ") + Chain(fileId) + Chain(" pageId ") + Chain(pageId);
		    throw Exception(EXLOC, msg, e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Lock sys page failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}
	
	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }
    
    _sysPageLock[i].semId=semId;    
    _sysPageLock[i].fileId = fileId;
    _sysPageLock[i].pageId = pageId;
    
    return _sysPageLock[i].lockId;
    
}

void CegoLockHandler::unlockSysPage(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _sysPageLock[i].lockId != lockId )
    {
	i++;
    }
    if (_sysPageLock[i].lockId == lockId)
    {
	int semId = _sysPageLock[i].semId;
	
	if ( _lockCount[semId] > 1 )
	{
	    _lockCount[semId]--;
	}
	else
	{
	    _lockArray[semId]->unlock();
	    _lockCount[semId]=0;
	}
	_sysPageLock[i].lockId = 0;
	_sysPageLock[i].semId = 0;
    }
    else
    {
	Chain msg = Chain("Invalid sys page lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

unsigned long CegoLockHandler::lockData(CegoObject::ObjectType type, int fileId, int pageId, enum LockMode lockMode)
{

    if ( type == CegoObject::INDEX || type == CegoObject::UINDEX || type == CegoObject::PINDEX )
    {
	return lockIndexPage(fileId, pageId, lockMode);
    }
    else if ( type == CegoObject::RBSEG )
    {
	return lockRBPage(fileId, pageId, lockMode);
    }
    return lockDataPage(fileId, pageId, lockMode);
}


void CegoLockHandler::unlockData(CegoObject::ObjectType type, unsigned long lockId)
{
    if ( type == CegoObject::INDEX || type == CegoObject::UINDEX || type == CegoObject::PINDEX )
    {
	return unlockIndexPage(lockId);
    }
    else if ( type == CegoObject::RBSEG )
    {
	return unlockRBPage(lockId);
    }
    return unlockDataPage(lockId);
}

unsigned long CegoLockHandler::lockDataPage(int fileId, int pageId, enum LockMode lockMode)
{

    int semId = _numRecSema 
	+ _numRBRecSema
	+ _numSysRecSema 
	+ _numSysPageSema 
	+ ( fileId * ( pageId + 1 ) )  % _numDataPageSema;

    int timeout = _pageLockTimeout;
    int numTries = _numLockTries;
    
    int i=0;
    while ( _dataPageLock[i].lockId != 0 && i < LCKMNG_NUMLOCKS)
    {
	i++;
    }    
    if (i == LCKMNG_NUMLOCKS)
    {
	throw Exception(EXLOC, Chain("Number of data page locks exceeded"));    
    }
    _dataPageLock[i].lockId = _nextLockId;    
    _nextLockId++;

    if ( _lockCount[semId] == 0 )
    {
	while ( numTries > 0 )
	{
	    try
	    {
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else 
		    _lockArray[semId]->writeLock(timeout);				
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;						
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock data page failed for fileId ") + Chain(fileId) + Chain(" pageId ") + Chain(pageId);
		    throw Exception(EXLOC, msg, e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Lock data page failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}
	
	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }

    _dataPageLock[i].semId=semId;    
    _dataPageLock[i].fileId = fileId;
    _dataPageLock[i].pageId = pageId;


    return _dataPageLock[i].lockId;
    
}

void CegoLockHandler::unlockDataPage(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _dataPageLock[i].lockId != lockId )
    {
	i++;
    }
    if (_dataPageLock[i].lockId == lockId)
    {
	int semId = _dataPageLock[i].semId;
	
	if ( _lockCount[semId] > 1 )
	{
	    _lockCount[semId]--;
	}
	else
	{
	    _lockArray[semId]->unlock();
	    _lockCount[semId]=0;
	}
	_dataPageLock[i].lockId = 0;
	_dataPageLock[i].semId = 0;
    }
    else
    {
	Chain msg = Chain("Invalid data page lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

unsigned long CegoLockHandler::lockIndexPage(int fileId, int pageId, enum LockMode lockMode)
{

    int semId = _numRecSema
	+ _numRBRecSema 
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ ( fileId * ( pageId + 1 ) )  % _numIdxPageSema;

    int timeout = _pageLockTimeout;
    int numTries = _numLockTries;

    int i=0;
    while ( _idxPageLock[i].lockId != 0 && i < LCKMNG_NUMLOCKS)
    {
	i++;
    }
    if (i == LCKMNG_NUMLOCKS)
    {
	throw Exception(EXLOC, "Number of index page locks exceeded");    
    }
    _idxPageLock[i].lockId = _nextLockId;    
    _nextLockId++;

    if ( _lockCount[semId] == 0 )
    {
	while ( numTries > 0 )
	{
	    try
	    {
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else 
		    _lockArray[semId]->writeLock(timeout);
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;									
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock index page failed for fileId ") + Chain(fileId) + Chain(" pageId ") + Chain(pageId);
		    throw Exception(EXLOC, msg, e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Lock index page failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}

	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }

    _idxPageLock[i].semId=semId;    
    _idxPageLock[i].fileId = fileId;
    _idxPageLock[i].pageId = pageId;


    return _idxPageLock[i].lockId;
    
}

void CegoLockHandler::unlockIndexPage(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _idxPageLock[i].lockId != lockId )
    {
	i++;
    }
    if (_idxPageLock[i].lockId == lockId)
    {
	int semId = _idxPageLock[i].semId;
	
	if ( _lockCount[semId] > 1 )
	{
	    _lockCount[semId]--;
	}
	else
	{
	    _lockArray[semId]->unlock();
	    _lockCount[semId]=0;
	}
	_idxPageLock[i].lockId = 0;
	_idxPageLock[i].semId = 0;
    }
    else
    {
	Chain msg = Chain("Invalid index page lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

unsigned long CegoLockHandler::lockRBPage(int fileId, int pageId, enum LockMode lockMode)
{

    int semId = _numRecSema 
	+ _numRBRecSema 
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema 
	+ ( fileId * ( pageId + 1 ) )  % _numRBPageSema;

    int timeout = _pageLockTimeout;
    int numTries = _numLockTries;

    int i=0;
    while ( _rbPageLock[i].lockId != 0 && i < LCKMNG_NUMLOCKS)
    {
	i++;
    }
    if (i == LCKMNG_NUMLOCKS)
    {
	throw Exception(EXLOC, "Number of rb page locks exceeded");    
    }
    _rbPageLock[i].lockId = _nextLockId;    
    _nextLockId++;

    if ( _lockCount[semId] == 0 )
    {

	while ( numTries > 0 )
	{
	    try
	    {
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else 
		    _lockArray[semId]->writeLock(timeout);
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock rb page failed for fileId ") + Chain(fileId) + Chain(" pageId ") + Chain(pageId);
		    throw Exception(EXLOC, msg, e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Lock rb page failed (SemId=") + Chain(semId) + Chain("), still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}

	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }

    _rbPageLock[i].semId=semId;    
    _rbPageLock[i].fileId = fileId;
    _rbPageLock[i].pageId = pageId;


    return _rbPageLock[i].lockId;
    
}

void CegoLockHandler::unlockRBPage(unsigned long lockId)
{

    int i=0;
    while (i < LCKMNG_NUMLOCKS && _rbPageLock[i].lockId != lockId )
    {
	i++;
    }
    if (_rbPageLock[i].lockId == lockId)
    {
	int semId = _rbPageLock[i].semId;
	
	if ( _lockCount[semId] > 1 )
	{
	    _lockCount[semId]--;
	}
	else
	{
	    _lockArray[semId]->unlock();
	    _lockCount[semId]=0;
	}
	_rbPageLock[i].lockId = 0;
	_rbPageLock[i].semId = 0;
    }
    else
    {
	Chain msg = Chain("Invalid rb page lockid ") + Chain(lockId);
	throw Exception(EXLOC, msg);
    }
}

void CegoLockHandler::lockDataFile(int fileId, enum LockMode lockMode)
{

    int semId = _numRecSema 
	+ _numRBRecSema 
	+ _numSysRecSema 
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema
	+ _numRBPageSema 
	+ fileId % _numDataFileSema;

    int timeout = _fileLockTimeout;
    int numTries = _numLockTries;

    if ( _lockCount[semId] == 0 )
    {

	while ( numTries > 0 )
	{
	    try
	    {
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else
		{
		    _lockArray[semId]->writeLock(timeout);
		}
		numTries=0;
	    }
	    catch ( Exception e )
	    {
		numTries--;
		if ( numTries == 0 )
		{
		    Chain msg = Chain("Lock data file failed for fileId ") + Chain(fileId);
		    throw Exception(EXLOC, msg, e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Lock data file failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}

	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }
    
}

void CegoLockHandler::unlockDataFile(int fileId)
{
    int semId = _numRecSema
	+ _numRBRecSema 
	+ _numSysRecSema 
	+ _numSysPageSema 
	+ _numDataPageSema
	+ _numIdxPageSema  
	+ _numRBPageSema
	+ fileId % _numDataFileSema;

    if ( _lockCount[semId] == 1 )
    {
	_lockArray[semId]->unlock();
	_lockCount[semId]--;
    }
    else
    {
	_lockCount[semId]--;
    }
}

void CegoLockHandler::lockBufferPool(int hashId, enum LockMode lockMode)
{

    int semId = _numRecSema
	+ _numRBRecSema 
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema 
	+ _numIdxPageSema	
	+ _numRBPageSema  
	+ _numDataFileSema 
	+ hashId % _numBufferPoolSema;

    int timeout = _poolLockTimeout;
    int numTries = _numLockTries;

    if ( _lockCount[semId] == 0 )
    {
	while ( numTries > 0 )
	{
	    try
	    {		
		if ( lockMode == READ )
		    _lockArray[semId]->readLock(timeout);
		else 
		    _lockArray[semId]->writeLock(timeout);				
		numTries=0;
	    }
	    catch ( Exception e )
	    {		
		numTries--;					
		if ( numTries == 0 )
		{
		    throw Exception(EXLOC, "Lock buffer pool failed", e);
		}
		else
		{
		    _pDBMng->log(_modId, Logger::NOTICE, Chain("Lock buffer pool failed, still ") + Chain(numTries) + Chain(" tries"));
		}
	    }
	}

	_lockCount[semId]++;
    }
    else
    {
	_lockCount[semId]++;
    }
    
}

void CegoLockHandler::unlockBufferPool(int hashId)
{

    int semId = _numRecSema 
	+ _numRBRecSema 
	+ _numSysRecSema
	+ _numSysPageSema 
	+ _numDataPageSema
	+ _numIdxPageSema
	+ _numRBPageSema   
	+ _numDataFileSema 
	+ hashId % _numBufferPoolSema;

    if ( _lockCount[semId] == 1 )
    {
	_lockArray[semId]->unlock();
	_lockCount[semId]--;
    }
    else
    {
	_lockCount[semId]--;
    }
}

void CegoLockHandler::lockBufferPool()
{

    for ( int i=0; i<_numBufferPoolSema; i++ )
	lockBufferPool(i, WRITE);
}

void CegoLockHandler::unlockBufferPool()
{

    for ( int i=0; i<_numBufferPoolSema; i++ )
	unlockBufferPool(i);
}

void CegoLockHandler::getLockStat(int semId, Chain& lockName, long& lockCount,  long &numRdLock, long &numWrLock, long &sumRdDelay, long &sumWrDelay)
{
    
    lockName = _lockArray[semId]->getId();
    lockCount = _lockArray[semId]->numLockTry();
    numRdLock = _lockArray[semId]->numReadLock();
    numWrLock = _lockArray[semId]->numWriteLock();
    sumRdDelay = 0;
    sumWrDelay = 0;

    if ( _lockArray[semId]->numReadLock() > 0 )
	sumRdDelay = _lockArray[semId]->sumReadDelay() / LCKMNG_DELRES;   
    if ( _lockArray[semId]->numWriteLock() > 0 )
	sumWrDelay = _lockArray[semId]->sumWriteDelay() / LCKMNG_DELRES;

}

void CegoLockHandler::getAggLockStat(const Chain& lockGroup, int& numLock, long& lockCount,  long &numRdLock, long &numWrLock, long &sumRdDelay, long &sumWrDelay)
{

    lockCount = 0;
    numRdLock = 0;
    numWrLock = 0;
    sumRdDelay = 0;
    sumWrDelay = 0;
    numLock = 0;

    for ( int semId=0; semId<_numSema; semId++ )
    {
	
	Tokenizer tok(_lockArray[semId]->getId(), "-");
	
	Chain checkGroup;
	tok.nextToken(checkGroup);
	
	if ( checkGroup == lockGroup )
	{
	    numLock++;
	    lockCount += _lockArray[semId]->numLockTry();
	    numRdLock += _lockArray[semId]->numReadLock();
	    numWrLock += _lockArray[semId]->numWriteLock();
	    
	    if ( _lockArray[semId]->numReadLock() > 0 )
		sumRdDelay += _lockArray[semId]->sumReadDelay() / LCKMNG_DELRES;   
	    if ( _lockArray[semId]->numWriteLock() > 0 )
		sumWrDelay += _lockArray[semId]->sumWriteDelay() / LCKMNG_DELRES;
	}
    }

}
