xref: /dports/databases/cego/cego-2.47.11/src/CegoDatabaseManager.cc

///////////////////////////////////////////////////////////////////////////////
//
// CegoDatabaseManager.cc
// ----------------------
// Cego Database Manager implementation
//
// Design and Implementation by Bjoern Lemke
//
// (C)opyright 2000-2019 Bjoern Lemke
//
// IMPLEMENTATION MODULE
//
// Class: CegoDatabaseManager
//
// Description: General database manager
//
// Status: CLEAN
//
///////////////////////////////////////////////////////////////////////////////

// LFC INCLUDES
#include <lfcbase/ThreadLock.h>
#include <lfcbase/File.h>
#include <lfcbase/Datetime.h>
#include <lfcbase/Sleeper.h>
#include <lfcbase/Net.h>
#include <lfcbase/NetHandler.h>
#include <lfcxml/XMLSuite.h>
#include <lfcxml/Element.h>
#include <lfcxml/Document.h>

// CEGO INCLUDES
#include "CegoDatabaseManager.h"
#include "CegoTypeConverter.h"
#include "CegoDefs.h"
#include "CegoXMLdef.h"
#include "CegoXMLSpace.h"

// POSIX INCLUDES
#include <string.h>
#include <stdlib.h>

static ThreadLock dbmLock("DBM");
extern bool __lockStatOn;

//////////////////
// ObjectRecord //
//////////////////

CegoDatabaseManager::ObjectRecord::ObjectRecord()
{
}

CegoDatabaseManager::ObjectRecord::ObjectRecord(const int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    _objName = objName;
    _type = type;
    _tabSetId = tabSetId;
    _numUsed = 0;
    _mode = SHARED;
    _tid = 0;
}

CegoDatabaseManager::ObjectRecord::~ObjectRecord()
{
}

const int CegoDatabaseManager::ObjectRecord::getTabSetId() const
{
    return _tabSetId;
}

void CegoDatabaseManager::ObjectRecord::incUsed()
{
    _numUsed++;
}

void CegoDatabaseManager::ObjectRecord::decUsed()
{
    if ( _numUsed > 0 )
	_numUsed--;
}

int CegoDatabaseManager::ObjectRecord::numUsed() const
{
    return _numUsed;
}

void CegoDatabaseManager::ObjectRecord::setMode(ObjectUseMode mode)
{
    _mode = mode;
}

CegoDatabaseManager::ObjectUseMode CegoDatabaseManager::ObjectRecord::getMode() const
{
    return _mode;
}

void CegoDatabaseManager::ObjectRecord::setTid(unsigned long long tid)
{
    _tid = tid;
}

unsigned long long CegoDatabaseManager::ObjectRecord::getTid() const
{
    return _tid;
}

const Chain& CegoDatabaseManager::ObjectRecord::getName() const
{
    return _objName;
}

const CegoObject::ObjectType CegoDatabaseManager::ObjectRecord::getType() const
{
    return _type;
}

CegoDatabaseManager::ObjectRecord& CegoDatabaseManager::ObjectRecord::operator = ( const CegoDatabaseManager::ObjectRecord& t)
{
    _tabSetId = t._tabSetId;
    _objName = t._objName;
    _type = t._type;
    _numUsed = t._numUsed;
    _mode = t._mode;
    _tid = t._tid;
    return (*this);
}

bool CegoDatabaseManager::ObjectRecord::operator == ( const CegoDatabaseManager::ObjectRecord& r)
{
    bool typeMatch=false;
    if ( ( _type == CegoObject::PAVLTREE || _type == CegoObject::UAVLTREE || _type == CegoObject::AVLTREE )
	 && ( r._type == CegoObject::PAVLTREE || r._type == CegoObject::UAVLTREE || r._type == CegoObject::AVLTREE ) )
    {
	typeMatch=true;
    }
    else if ( ( _type == CegoObject::PBTREE || _type == CegoObject::UBTREE || _type == CegoObject::BTREE )
	&& ( r._type == CegoObject::PBTREE || r._type == CegoObject::UBTREE || r._type == CegoObject::BTREE ) )
    {
	typeMatch=true;
    }
    else
    {
	typeMatch = _type == r._type;
    }

    if ( _tabSetId == r._tabSetId && _objName == r._objName && typeMatch )
	return true;
    return false;
}

//////////////////
// CopyRecord //
//////////////////

CegoDatabaseManager::CopyRecord::CopyRecord()
{
}

CegoDatabaseManager::CopyRecord::CopyRecord(const Chain& tableSet, const Chain& targetHost, const Chain& mediatorHost, const Chain& user, const Chain& passwd, const Chain& msg)
{
    _id = 0;
    _tableSet = tableSet;
    _targetHost = targetHost;
    _mediatorHost = mediatorHost;
    _user = user;
    _passwd = passwd;
    _msg = msg;
}

CegoDatabaseManager::CopyRecord::~CopyRecord()
{
}

const Chain& CegoDatabaseManager::CopyRecord::getTableSet() const
{
    return _tableSet;
}

const Chain& CegoDatabaseManager::CopyRecord::getTargetHost() const
{
    return _targetHost;
}

const Chain& CegoDatabaseManager::CopyRecord::getMediatorHost() const
{
    return _mediatorHost;
}

const Chain& CegoDatabaseManager::CopyRecord::getUser() const
{
    return _user;
}

const Chain& CegoDatabaseManager::CopyRecord::getPasswd() const
{
    return _passwd;
}

void CegoDatabaseManager::CopyRecord::setMsg(const Chain& msg)
{
    _msg = msg;
}

const Chain& CegoDatabaseManager::CopyRecord::getMsg() const
{
    return _msg;
}

int CegoDatabaseManager::CopyRecord::getId() const
{
    return _id;
}

void CegoDatabaseManager::CopyRecord::setId(int id)
{
    _id = id;
}

CegoDatabaseManager::CopyRecord& CegoDatabaseManager::CopyRecord::operator = ( const CegoDatabaseManager::CopyRecord& cr)
{
    _id = cr._id;
    _tableSet = cr._tableSet;
    _targetHost = cr._targetHost;
    _mediatorHost = cr._mediatorHost;

    _user = cr._user;
    _passwd = cr._passwd;
    _msg = cr._msg;

    return (*this);
}

bool CegoDatabaseManager::CopyRecord::operator == ( const CegoDatabaseManager::CopyRecord& cr)
{
    if ( _tableSet == cr._tableSet && _targetHost == cr._targetHost)
	return true;
    return false;
}

/////////////////////
// DbSessionRecord //
/////////////////////

CegoDatabaseManager::DbSessionRecord::DbSessionRecord()
{
}

CegoDatabaseManager::DbSessionRecord::DbSessionRecord(CegoDistDbHandler* pHandler)
{
    _pHandler = pHandler;
}

CegoDatabaseManager::DbSessionRecord::DbSessionRecord(const Chain& hostName, const Chain& tableSet,
				     const Chain& userName, CegoDistDbHandler* pHandler)
{
    _hostName = hostName;
    _tableSet = tableSet;
    _userName = userName;
    _pHandler = pHandler;
    _lastUsed = Datetime().asLong();
}

CegoDatabaseManager::DbSessionRecord::~DbSessionRecord()
{
}

const Chain& CegoDatabaseManager::DbSessionRecord::getHostName() const
{
    return _hostName;
}

const Chain& CegoDatabaseManager::DbSessionRecord::getTableSet() const
{
    return _tableSet;
}

const Chain& CegoDatabaseManager::DbSessionRecord::getUserName() const
{
    return _userName;
}

CegoDistDbHandler* CegoDatabaseManager::DbSessionRecord::getDbHandler() const
{
    return _pHandler;
}

bool CegoDatabaseManager::DbSessionRecord::isUsed() const
{
    return _isUsed;
}
void CegoDatabaseManager::DbSessionRecord::setUsed(bool isUsed)
{
    _isUsed = isUsed;
}

unsigned long long CegoDatabaseManager::DbSessionRecord::getTSLastUsed()
{
    return _lastUsed;
}

void CegoDatabaseManager::DbSessionRecord::setTSLastUsed(unsigned long long ts)
{
    _lastUsed = ts;
}

CegoDatabaseManager::DbSessionRecord& CegoDatabaseManager::DbSessionRecord::operator = ( const CegoDatabaseManager::DbSessionRecord& sr)
{
    _hostName = sr._hostName;
    _tableSet = sr._tableSet;
    _userName = sr._userName;
    _pHandler = sr._pHandler;
    _lastUsed = sr._lastUsed;
    return (*this);

}

bool CegoDatabaseManager::DbSessionRecord::operator == ( const CegoDatabaseManager::DbSessionRecord& sr)
{
    if ( _pHandler == sr._pHandler )
	return true;
    return false;
}

/////////////////////////
// CegoDatabaseManager //
/////////////////////////

CegoDatabaseManager::CegoDatabaseManager(const Chain& xmlDef, const Chain& lckFileName, const Chain& logFile, const Chain& progName, CegoDbHandler::ProtocolType protType) :  CegoBufferPool(xmlDef, logFile, progName)
{
    _protType = protType;

    _lckFileName = lckFileName;

    File lckFile(_lckFileName);
    if ( lckFile.exists() )
    {
	Chain msg = Chain("Running database instance detected at lock file ") + lckFileName;
	throw Exception(EXLOC, msg);
    }
    else
    {
	lckFile.open(File::WRITE);
	lckFile.writeChain("Locked");
	_nextBeat=0;
	lckFile.close();
    }

    dbmLock.init(LCKMNG_LOCKWAITDELAY, __lockStatOn);
    for ( int i=0; i<TABMNG_MAXTABSET; i++)
    {
	_recoveryMode[i] = OFF;
	_pQueryCache[i] = 0;
	_pTableCache[i]= 0;
    }
    _nextCopyId=1;

    _logConfigured=false;

    _modId = getModId("CegoDatabaseManager");
}

CegoDatabaseManager::~CegoDatabaseManager()
{
    File lckFile(_lckFileName);
    lckFile.remove();

    for ( int i=0; i<TABMNG_MAXTABSET; i++)
    {
	if ( _pQueryCache[i] )
	    delete _pQueryCache[i];
	if ( _pTableCache[i] )
	    delete _pTableCache[i];
    }
}

void CegoDatabaseManager::allocateQueryCache(const Chain& tableSet)
{
    if ( getQueryCacheMode(tableSet) )
    {
	int maxEntry = getMaxQueryCacheEntry(tableSet);
	int maxSize = getMaxQueryCacheSize(tableSet);
	int hashRange = getQueryCacheHashRange(tableSet);

	if ( maxEntry > 0 && maxSize > 0 && hashRange > 0)
	{
	    int tabSetId = getTabSetId(tableSet);

	    if ( _pQueryCache[tabSetId] )
		delete _pQueryCache[tabSetId];

	    _pQueryCache[tabSetId] = new CegoQueryCache(maxEntry, maxSize, hashRange);
	}
	else
	{
	    log(_modId, Logger::NOTICE, Chain("Query Cache Size/Entry not appropriate, skipping cache allocation"));
	}
    }
}

void CegoDatabaseManager::allocateTableCache(const Chain& tableSet)
{
    if ( getTableCacheMode(tableSet) )
    {

	int numEntry = getMaxTableCacheEntry(tableSet);
	int cacheSize = getMaxTableCacheSize(tableSet);

	if ( numEntry > 0 && cacheSize > 0 )
	{
	    int tabSetId = getTabSetId(tableSet);

	    if ( _pTableCache[tabSetId] )
		delete _pTableCache[tabSetId];
	    _pTableCache[tabSetId] = new CegoTableCache(numEntry,cacheSize, this);
	}
	else
	{
	    log(_modId, Logger::NOTICE, Chain("Table Cache Size/Entry not appropriate, skipping cache allocation"));
	}
    }
}

void CegoDatabaseManager::releaseQueryCache(const Chain& tableSet)
{
    int tabSetId = getTabSetId(tableSet);

    if ( _pQueryCache[tabSetId] )
    {
	delete _pQueryCache[tabSetId];
	_pQueryCache[tabSetId]=0;
    }
}

void CegoDatabaseManager::releaseTableCache(const Chain& tableSet)
{
    int tabSetId = getTabSetId(tableSet);

    if ( _pTableCache[tabSetId] )
    {
	delete _pTableCache[tabSetId];
	_pTableCache[tabSetId]=0;
    }
}

void CegoDatabaseManager::cleanCache(int tabSetId, CegoObject::ObjectType objType, const Chain& objName)
{
    if ( _pQueryCache[tabSetId] )
    {
	_pQueryCache[tabSetId]->invalidate(CegoObject(objType, objName, tabSetId));
    }

    if ( _pTableCache[tabSetId] && objType == CegoObject::TABLE )
    {
	_pTableCache[tabSetId]->invalidate(tabSetId, objName);
    }
}

CegoQueryCache* CegoDatabaseManager::getQueryCache(const Chain& tableSet) const
{
    int tabSetId = getTabSetId(tableSet);
    return _pQueryCache[tabSetId];
}

CegoQueryCache* CegoDatabaseManager::getQueryCache(int tabSetId) const
{
    return _pQueryCache[tabSetId];
}

CegoTableCache* CegoDatabaseManager::getTableCache(const Chain& tableSet) const
{
    int tabSetId = getTabSetId(tableSet);
    return _pTableCache[tabSetId];
}

CegoTableCache* CegoDatabaseManager::getTableCache(int tabSetId) const
{
    return _pTableCache[tabSetId];
}

void CegoDatabaseManager::beat()
{
    File lckFile(_lckFileName);
    lckFile.open(File::WRITE);
    lckFile.writeChain(Chain("Beat=") + Chain(_nextBeat) + Chain("\n"));
    _nextBeat++;
    lckFile.close();
}


void CegoDatabaseManager::checkTableSetRunState(int tabSetId)
{
    Chain runState = getTableSetRunState(tabSetId);

    if ( runState != Chain(XML_ONLINE_VALUE)
	 && runState != Chain(XML_BACKUP_VALUE)
	 && runState != Chain(XML_RECOVERY_VALUE)
	 && runState != Chain(XML_CHECKPOINT_VALUE) )
    {
	Chain msg = Chain("Tableset ") + getTabSetName(tabSetId) + Chain(" not online ( run state is ") + runState + Chain(" )");
	throw Exception(EXLOC, msg);
    }
}


void CegoDatabaseManager::PR()
{
    dbmLock.readLock(DBM_LOCKTIMEOUT);
}

void CegoDatabaseManager::PW()
{
    dbmLock.writeLock(DBM_LOCKTIMEOUT);
}

void CegoDatabaseManager::V()
{
    dbmLock.unlock();
}

bool CegoDatabaseManager::isLoggerConfigured()
{
    return _logConfigured;
}

CegoDatabaseManager::RecoveryMode CegoDatabaseManager::getRecoveryMode(int tabSetId)
{
    return _recoveryMode[tabSetId];
}

void CegoDatabaseManager::setRecoveryMode(int tabSetId, CegoDatabaseManager::RecoveryMode m)
{
    _recoveryMode[tabSetId] = m;
}

void CegoDatabaseManager::setAllRecoveryOff()
{
    for ( int i=0; i<TABMNG_MAXTABSET; i++)
    {
	_recoveryMode[i] = OFF;
    }
}

void CegoDatabaseManager::startRecovery(const Chain& tableSet)
{
    PW();
    _recoveryList.Insert(tableSet);
    V();
}

void CegoDatabaseManager::startCopy(const Chain& tableSet, const Chain& targetHost, const Chain& mediatorHost, const Chain& user, const Chain& passwd, const Chain& msg)
{
    PW();
    _copyList.Insert( CopyRecord( tableSet, targetHost, mediatorHost, user, passwd, msg));
    V();
}

bool CegoDatabaseManager::nextRecovery(Chain& tableSet)
{
    PW();
    Chain *pS = _recoveryList.First();
    if ( pS )
    {
	tableSet = *pS;
	_recoveryList.Remove(tableSet);
	V();
	return true;
    }
    V();
    return false;
}

bool CegoDatabaseManager::nextCopy(int& id, Chain& tableSet, Chain& targetHost, Chain& mediatorHost, Chain& user, Chain& passwd)
{
    PW();
    CopyRecord *pCR = _copyList.First();
    while ( pCR )
    {
	if ( pCR->getId() == 0 )
	{
	    id = _nextCopyId++;
	    pCR->setId(id);

	    tableSet = pCR->getTableSet();
	    targetHost = pCR->getTargetHost();
	    mediatorHost = pCR->getMediatorHost();
	    user = pCR->getUser();
	    passwd = pCR->getPasswd();

	    // _copyList.Remove(*pCR);
	    V();
	    return true;
	}
	pCR = _copyList.Next();
    }
    V();
    return false;
}

void CegoDatabaseManager::setCopyStatus(int id, const Chain& msg)
{
    PW();
    CopyRecord *pCR = _copyList.First();
    while ( pCR )
    {
	if ( pCR->getId() == id )
	{
	    pCR->setMsg(msg);

	    V();
	    return;
	}
	pCR = _copyList.Next();
    }
    V();
    return;
}

Element* CegoDatabaseManager::getCopyInfo()
{
    Element* pCopyInfo = new Element(XML_COPYINFO_ELEMENT);

    PR();
    CopyRecord *pCR = _copyList.First();
    while ( pCR )
    {

	Element *pN = new Element(XML_COPY_ELEMENT);

	pN->setAttribute(XML_CID_ATTR, pCR->getId());
	pN->setAttribute(XML_HOSTNAME_ATTR, pCR->getTargetHost());
	pN->setAttribute(XML_TABLESET_ATTR, pCR->getTableSet());
	pN->setAttribute(XML_STATUS_ATTR, pCR->getMsg());

	pCopyInfo->addContent(pN);

	pCR = _copyList.Next();
    }

    V();

    return pCopyInfo;
}

void CegoDatabaseManager::useObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type, ObjectUseMode mode, unsigned long long tid)
{
    // In SHARED mode, an object can be use in parallel by several threads. Normally, this is done, if the object is accessed read only.

    // we fist check, if corresponding tableset is really not offline
    checkTableSetRunState(tabSetId);


    /*
    Chain runState = getTableSetRunState(tabSetId);
    if ( runState != Chain(XML_ONLINE_VALUE) && runState != Chain(XML_BACKUP_VALUE) && runState != Chain(XML_CHECKPOINT_VALUE) )
    {
	Chain msg = Chain("Tableset ") + getTabSetName(tabSetId) + Chain(" not online ( run state is ") + runState + Chain(" )");
	throw Exception(EXLOC, msg);
    }
    */

    if ( mode == SHARED )
    {

        PW();
        ObjectRecord *pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));
        if ( pOR == 0 )
        {
            V();
	    Chain msg = Chain("Cannot access object <") + objName + Chain(">");
            throw Exception(EXLOC, msg);
        }

	if ( pOR->numUsed() > MAX_OBJECT_USAGE )
	{
            V();
	    Chain msg = Chain("Usage exceeded for <") + objName + Chain(">");
            throw Exception(EXLOC, msg);
	}

	if ( pOR->getTid() != 0 && pOR->getTid() == tid )
	{
	    pOR->incUsed();
	    V();
	}
	else
	{
	    int numTries=0;

	    while ( pOR && pOR->getMode() == EXCLUSIVE_WRITE && numTries < DBM_MAXLOCKTRIES )
	    {
		V();

		if ( numTries > 0 )
		{
		    if ( numTries > 1 )
		    {
			log(_modId, Logger::NOTICE, Chain("Repeated shared lock delay on ") + objName + Chain(" ( ") + Chain(numTries) + Chain(" tries )"));
		    }

		    Sleeper ns;
		    int sec = DBM_LOCKDELAY / 1000;
		    if ( sec > 0 )
			ns.secSleep(sec);
		    int msec = DBM_LOCKDELAY % 1000;
		    if ( msec > 0 )
			ns.milliSleep(msec);
		}

		PW();

		pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));

		if ( pOR == 0 )
		{
		    V();
		    Chain msg = Chain("Cannot access object ") + objName;
		    throw Exception(EXLOC, msg);
		}
		numTries++;
	    }

	    if ( numTries == DBM_MAXLOCKTRIES )
	    {
		V();
		Chain msg = Chain("Access timeout on object ") + objName;
		throw Exception(EXLOC, msg);
	    }

	    pOR->incUsed();
	    V();
	}
    }

    // In EXCLUSIVE_WRITE mode, an object is accessed exclusively by one thread only

    else if ( mode == EXCLUSIVE_WRITE )
    {
        PW();
        ObjectRecord *pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));
        if ( pOR == 0 )
        {
            V();
            Chain msg = Chain("Cannot access object ") + objName;
            throw Exception(EXLOC, msg);
        }

	if ( pOR->getTid() != 0 && pOR->getTid() == tid )
	{
	    pOR->incUsed();
	    V();
	}
	else
	{
	    if ( pOR->getMode() == SHARED && pOR->numUsed() == 0 )
	    {
		pOR->setMode(mode);
		pOR->setTid(tid);
	    }
	    else
	    {
		int numTries=0;
		while ( pOR && pOR->numUsed() != 0 && numTries < DBM_MAXLOCKTRIES )
		{

		    V();

		    if ( numTries > 0 )
		    {
			if ( numTries > 1 )
			{
			    log(_modId, Logger::NOTICE, Chain("Repeated exclusive write lock delay on ") + objName + Chain(" ( ") + Chain(numTries) + Chain(" tries )"));
			}

			Sleeper ns;

			int sec = DBM_LOCKDELAY / 1000;
			if ( sec > 0 )
			    ns.secSleep(sec);
			int msec = DBM_LOCKDELAY % 1000;
			if ( msec > 0 )
			    ns.milliSleep(msec);
		    }

		    PW();

		    pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));
		    if ( pOR == 0 )
		    {
			V();
			Chain msg = Chain("Cannot access object ") + objName;
			throw Exception(EXLOC, msg);
		    }

		    numTries++;

		}

		if ( numTries == DBM_MAXLOCKTRIES )
		{
		    V();
		    Chain msg = Chain("Access timeout on object ") + objName;
		    throw Exception(EXLOC, msg);
		}

		pOR->setMode(mode);
		pOR->setTid(tid);

	    }

	    pOR->incUsed();
	    V();
	}
    }
    return;
}

void CegoDatabaseManager::unuseObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type, unsigned long long tid )
{

    // we fist check, if corresponding tableset is really not offline
    checkTableSetRunState(tabSetId);

    /*
    Chain runState = getTableSetRunState(tabSetId);
    if ( runState != Chain(XML_ONLINE_VALUE) && runState != Chain(XML_BACKUP_VALUE) && runState != Chain(XML_CHECKPOINT_VALUE) )
    {
	Chain msg = Chain("Tableset ") + getTabSetName(tabSetId) + Chain(" not online ( run state is ") + runState + Chain(" )");
	throw Exception(EXLOC, msg);
    }
    */


    PW();
    ObjectRecord *pOR = _objList.Find(ObjectRecord(tabSetId, objName, type));

    if ( pOR == 0 )
    {
        V();
	Chain msg = Chain("Cannot access object ") + objName;
	throw Exception(EXLOC, msg);
	return;
    }

    // we just unuse object, if tid not specified or lockmode of object is SHARED or if transaction owns object lockmode of obejct id EXCLUSIVE_WRITE
    if ( ( tid == pOR->getTid() && pOR->getMode() == EXCLUSIVE_WRITE ) || tid == 0 || pOR->getMode() == SHARED )
    {
	pOR->decUsed();

	if ( pOR->numUsed() == 0 )
	{
	    pOR->setMode(SHARED);
	    pOR->setTid(0);
	}
    }
    else
    {
        V();
	Chain msg = Chain("Table ") + objName + Chain(" not owned by thread");
	throw Exception(EXLOC, msg);
	return;
    }

    V();
}

void CegoDatabaseManager::addObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    PW();
    _objList.Insert(ObjectRecord(tabSetId, objName, type));
    V();
}

void CegoDatabaseManager::removeObject(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{
    PW();
    _objList.Remove(ObjectRecord(tabSetId, objName, type));
    V();
}

void CegoDatabaseManager::removeAllObjects(int tabSetId)
{
    PW();
    ObjectRecord *pOR = _objList.First();
    while ( pOR )
    {
	if ( pOR->getTabSetId() == tabSetId )
	{
	    _objList.Remove(*pOR);
	    pOR = _objList.First();
	}
	else
	{
	    pOR = _objList.Next();
	}
    }
    V();
}

bool CegoDatabaseManager::objectExists(int tabSetId, const Chain& objName, CegoObject::ObjectType type)
{

    // we fist check, if corresponding tableset is really not offline
    checkTableSetRunState(tabSetId);

    /*
    Chain runState = getTableSetRunState(tabSetId);
    if ( runState != Chain(XML_ONLINE_VALUE) && runState != Chain(XML_BACKUP_VALUE) && runState != Chain(XML_CHECKPOINT_VALUE) )
    {
	Chain msg = Chain("Tableset ") + getTabSetName(tabSetId) + Chain(" not online ( run state is ") + runState + Chain(" )");
	throw Exception(EXLOC, msg);
    }
    */

    PR();
    ObjectRecord *pTR = _objList.Find(ObjectRecord(tabSetId, objName, type));
    V();
    if ( pTR )
    {
	return true;
    }
    else
    {
	return false;
    }
}

void CegoDatabaseManager::printObjectList()
{
    PW();
    ObjectRecord *pOR = _objList.First();
    while ( pOR )
    {
	cout << "ObjListEntry : " << pOR->getName() << " Type = " << pOR->getType() << endl;
	pOR = _objList.Next();
    }
    V();
}

void CegoDatabaseManager::setThreadInfo(int numDbThread, int numAdmThread, int numLogThread)
{
    _numDbThread = numDbThread;
    _numAdmThread = numAdmThread;
    _numLogThread = numLogThread;
    _activeDbThread = 0;
    _activeAdmThread = 0;
    _activeLogThread = 0;
}

void CegoDatabaseManager::getThreadInfo(int& numDbThread, int& numAdmThread, int& numLogThread,
		   int& activeDbThread, int& activeAdmThread, int& activeLogThread)
{
    numDbThread = _numDbThread;
    numAdmThread = _numAdmThread;
    numLogThread = _numLogThread;
    activeDbThread = _activeDbThread;
    activeAdmThread = _activeAdmThread;
    activeLogThread = _activeLogThread;
}

void CegoDatabaseManager::increaseActiveAdmThread()
{
    _activeAdmThread++;
}

void CegoDatabaseManager::decreaseActiveAdmThread()
{
    _activeAdmThread--;
}

void CegoDatabaseManager::increaseActiveDbThread()
{
    _activeDbThread++;
}

void CegoDatabaseManager::decreaseActiveDbThread()
{
    _activeDbThread--;
}

void CegoDatabaseManager::increaseActiveLogThread()
{
    _activeLogThread++;
}

void CegoDatabaseManager::decreaseActiveLogThread()
{
    _activeLogThread--;
}

void CegoDatabaseManager::getDBMLockStat(Chain& lockName, int& lockCount, unsigned long long &numRdLock, unsigned long long &numWrLock, unsigned long long &sumRdDelay, unsigned long long &sumWrDelay)
{
    lockName = dbmLock.getId();
    lockCount= dbmLock.numLockTry();

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

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

CegoDistDbHandler* CegoDatabaseManager::allocateSession(const Chain& hostName, const Chain& tableSet,
							const Chain& userName, const Chain& password)
{
    PW();

    CegoDistDbHandler* pSession = 0;

    try
    {
	DbSessionRecord *pSR = _dbSessionList.First();
	while ( pSR )
	{
	    if ( pSR->getHostName() == hostName
		 && pSR->getTableSet() == tableSet
		 && pSR->getUserName() == userName
		 && pSR->isUsed() == false )
	    {
		pSR->setUsed(true);
		Datetime dt;
		pSR->setTSLastUsed( dt.asLong() );

		V();

		return pSR->getDbHandler();
	    }
	    pSR = _dbSessionList.Next();
	}

	// create new session

	pSession = createSession(hostName, tableSet, userName, password);
	_dbSessionList.Insert( DbSessionRecord(hostName, tableSet, userName, pSession));

    }
    catch ( Exception e )
    {
	V();
	throw Exception(EXLOC, "Cannot allocate db session", e);
    }

    V();
    return pSession;
}

void CegoDatabaseManager::releaseSession(CegoDistDbHandler* pHandler)
{
    PW();

    DbSessionRecord *pSR = _dbSessionList.First();
    while ( pSR )
    {
	if ( pSR->getDbHandler() == pHandler  )
	{
	    pSR->setUsed(false);

	    V();

	    return;
	}
	pSR = _dbSessionList.Next();
    }

    V();

    Chain msg = Chain("Cannot release session for unknown db handle");
    throw Exception(EXLOC, msg);
}

void CegoDatabaseManager::cleanSession(int lifetime)
{
    PW();

    Datetime dt;

    DbSessionRecord *pSR = _dbSessionList.First();
    while ( pSR )
    {
	if ( pSR->getTSLastUsed() <  ( dt.asLong() - lifetime )  )
	{
	    if ( pSR->isUsed() == false )
	    {
		closeSession(pSR->getDbHandler());
		_dbSessionList.Remove( DbSessionRecord(pSR->getDbHandler()));
		pSR = _dbSessionList.First();
	    }
	}
	pSR = _dbSessionList.Next();
    }

    V();
}

Element* CegoDatabaseManager::getSessionInfo(int lifetime)
{
    Element* pSessionInfo = new Element(XML_DBSESSIONINFO_ELEMENT);

    DbSessionRecord *pSR = _dbSessionList.First();
    while ( pSR )
    {
	Element *pN = new Element(XML_DBSESSION_ELEMENT);
	pN->setAttribute(XML_HOSTNAME_ATTR, pSR->getHostName());
	pN->setAttribute(XML_TABLESET_ATTR, pSR->getTableSet());
	pN->setAttribute(XML_USER_ATTR, pSR->getUserName());
	if ( pSR->isUsed() )
	    pN->setAttribute(XML_ISUSED_ATTR, XML_TRUE_VALUE);
	else
	    pN->setAttribute(XML_ISUSED_ATTR, XML_FALSE_VALUE);

	Datetime dt;
	pN->setAttribute(XML_TTL_ATTR, Chain(pSR->getTSLastUsed() + (unsigned long long)lifetime - dt.asLong()));

	pSessionInfo->addContent(pN);

	pSR = _dbSessionList.Next();
    }
    return pSessionInfo;
}

CegoDistDbHandler* CegoDatabaseManager::createSession(const Chain& hostName, const Chain& tableSet,
						  const Chain& userName, const Chain& password)
{
    int portNo;
    getDataPort(portNo);

    Net n( NETMNG_MSG_BUFLEN, NETMNG_SIZEBUFLEN, getMaxSendLen() );
    NetHandler* pN = 0;
    CegoDistDbHandler *pSH = 0;

    try
    {

#ifdef DEBUG
	log(_modId, Logger::DEBUG, Chain("Connecting to ") + Chain(hostName) + Chain(" on port ") + Chain(portNo));
#endif

	pN = n.connect(hostName, portNo);
	pSH = new CegoDistDbHandler(pN, _protType, this);

#ifdef DEBUG
	log(_modId, Logger::DEBUG, Chain("Using user ") + userName + Chain("/") + password);
#endif
	pSH->requestSession(tableSet, userName, password, false);

    }
    catch ( Exception e )
    {

	if ( pSH )
	{
	    delete pSH;
	}

	if ( pN )
	{
	    delete pN;
	}

	Chain msg = Chain("Cannot create session to ") + hostName;
	throw Exception(EXLOC, msg);
    }

    return pSH;
}

void CegoDatabaseManager::closeSession(CegoDistDbHandler* pSH)
{
#ifdef DEBUG
    log(_modId, Logger::DEBUG, Chain("Closing session ..."));
#endif

    pSH->closeSession();

    NetHandler *pN = pSH->getNetHandler();

    delete pSH;
    delete pN;
}

void CegoDatabaseManager::configureLogger(Logger::LogLevel level)
{
    for ( int i=1; i< getMapSize() ; i++)
    {
	logModule(i, getModName(i), level);
    }
    _logConfigured=true;
}

void CegoDatabaseManager::configureLogger()
{
    ListT<Chain> modList;
    _logConfigured =  getModuleList(modList);

    Chain *pMod = modList.First();
    while ( pMod )
    {
	if ( (Chain)pMod->toUpper() == Chain("ALL"))
	{
	    Logger::LogLevel level = getLogLevel(*pMod);
	    for ( int i=1; i< getMapSize() ; i++)
	    {
		logModule(i, getModName(i), level);
	    }
	}
	else
	{
	    unsigned long modId = getModId(*pMod);
	    logModule(modId, *pMod, getLogLevel(*pMod));
	}
	pMod = modList.Next();
    }
}

bool CegoDatabaseManager::verifyJDBC(const Chain& user)
{
    SetT<Chain> roleSet;
    getRoleSet(user, roleSet);
    return roleSet.Find(Chain(ROLE_JDBC));
}

bool CegoDatabaseManager::verifyAccess(const int tabSetId, const Chain& objName, CegoObject::ObjectType type, CegoXMLSpace::AccessMode mode, const Chain& user)
{
    SetT<Chain> roleSet;

    getRoleSet(user, roleSet);
    Chain tableSet = getTabSetName(tabSetId);

    Chain *pRole = roleSet.First();
    while ( pRole )
    {

	Chain objPattern = objName;

	if ( matchRole( *pRole, tableSet, objPattern, mode ) )
	    return true;
	pRole = roleSet.Next();
    }
    return false;
}

void CegoDatabaseManager::initLogFiles(const Chain& tableSet, bool overwrite)
{
    ListT<Chain> lfList;
    ListT<int> sizeList;
    ListT<Chain> statusList;

    int tabSetId = getTabSetId(tableSet);
    getLogFileInfo(tableSet, lfList, sizeList, statusList);

    Chain *pLog = lfList.First();
    int *pSize = sizeList.First();

    bool isFirst = true;
    while ( pLog )
    {

	if ( isFirst )
	    setLogFileStatus(tableSet, *pLog, XML_ACTIVE_VALUE);
	else
	    setLogFileStatus(tableSet, *pLog,  XML_FREE_VALUE);

	isFirst=false;

	log(_modId, Logger::NOTICE, Chain("Initializing logfile ") + *pLog + Chain(" ..."));

	if ( overwrite == false )
	{
	    File checkLog(*pLog);
	    if ( checkLog.exists() )
	    {
		Chain msg = Chain("Cannot initialize logfile <") + *pLog + Chain(">, file already exists");
		throw Exception(EXLOC, msg);
	    }
	}

	setLogFile(tabSetId, *pLog, false);
	initLog(tabSetId, *pSize);

	pLog = lfList.Next();
	pSize = sizeList.Next();

    }
}

void CegoDatabaseManager::releaseLogFiles(const Chain& tableSet, bool waitForArch)
{
    ListT<Chain> lfList;
    ListT<int> sizeList;
    ListT<Chain> statusList;

    int tabSetId = getTabSetId(tableSet);
    getLogFileInfo(tableSet, lfList, sizeList, statusList);

    Chain *pLog = lfList.First();
    Chain *pStatus = statusList.First();

    while ( pLog && pStatus )
    {
        if ( *pStatus == Chain(XML_ACTIVE_VALUE) && File::exists(*pLog) )
        {
            setLogFile(tabSetId, *pLog, true);
            unsigned long long minlsn = getMinLSN(tabSetId);
            if ( minlsn > 0 )
            {
                log(_modId, Logger::NOTICE, Chain("Releasing logfile ") + *pLog + Chain(" LSN=") + Chain(minlsn));
                setLogFileStatus(tableSet, *pLog, XML_OCCUPIED_VALUE);
            }
        }
        pStatus = statusList.Next();
        pLog = lfList.Next();
    }

    if ( waitForArch )
    {
	bool notArchived = true;

	while ( notArchived )
	{

	    log(_modId, Logger::NOTICE, Chain("Waiting for archive ... "));

	    ListT<Chain> lfList;
	    ListT<int> sizeList;
	    ListT<Chain> statusList;

	    getLogFileInfo(tableSet, lfList, sizeList, statusList);

	    notArchived = false;

	    Chain *pStatus = statusList.First();
            Chain *pLog = lfList.First();
            while ( pLog && pStatus )
            {
                if ( *pStatus != Chain(XML_FREE_VALUE) && File::exists(*pLog) )
                    notArchived = true;
                pStatus = statusList.Next();
                pLog = lfList.Next();
            }

	    lfList.Empty();
	    sizeList.Empty();
	    statusList.Empty();

	    Sleeper s;
	    s.secSleep(LOGMNG_RECOVERY_DELAY);
	}
    }
}

CegoDbHandler::ProtocolType CegoDatabaseManager::getProtType() const
{
    return _protType;
}