Bdb packages | Design docs | Source docs | Guidelines | Recent releases
	Search | Site Map .

                 

---

/BdbClusteringServer/BdbSrvUtil.cc

Go to the documentation of this file.

//------------------------------------------------------------------------------
// File and Version Information:
//      $Id: BdbSrvUtil.cc,v 1.19 2002/08/30 01:13:31 ryd Exp $
//
// Description:
//      Transient class
//      A set of utility functions used by the clustering server
//
// Environment:
//      Software developed for the BaBar Detector at the SLAC B-Factory
//
// Author List:
//      Jacek Becla                Original Author
//
// Copyright Information:
//      Copyright (C) 2000      Stanford Linear Accelerator Center
//
//------------------------------------------------------------------------------



#include "BdbClusteringServer/BdbSrvUtil.hh"


#include "BdbUtil/BdbMutex.hh"
#include "BdbApplication/BdbDomain.hh"
#include "BdbAccess/BdbClustHintSetupMgr.hh"
#include "BdbClusteringServer/BdbActiveDbSpec.hh"
#include "BdbClusteringServer/BdbActiveDbsMgr.hh"
#include "BdbClusteringServer/BdbClSrvCRef.hh"
#include "BdbClusteringServer/BdbDbMetadataMgr.hh"
#include "BdbClusteringServer/BdbFSSelector.hh"
#include "BdbClusteringServer/BdbGroupDescr.hh"
#include "BdbClusteringServer/BdbSrvLogMgr.hh"


#include <iomanip.h>
#include <iostream.h>
#include <strstream.h>

#include <set>
using std::set;
#include <map>
using std::map;
#include <string>
using std::string;
#include <vector>
using std::vector;

extern ooStatus BaBarObjyDummyErrorHandler(ooErrorLevel, ooError&, ooHandle(ooObj)*, char*);

extern "C" {
#include <unistd.h>     /* access */
}


unsigned long     BdbSrvUtil::_reqID           = 0;
long              BdbSrvUtil::_lastReqTime     = 0;
BdbClustHintSetupMgr* BdbSrvUtil::_chsMgr      = 0;
BdbDbMetadataMgr* BdbSrvUtil::_dbmMgr          = 0;
BdbMutex          BdbSrvUtil::_chsMutex("chsMgrMutex");
BdbFSSelector*    BdbSrvUtil::_fsSelector      = 0;
ooContext*        BdbSrvUtil::_exclLockCtx     = 0;
ooTrans           BdbSrvUtil::_exclLockTrans;
char*             BdbSrvUtil::_logDir          = 0;
ooMode            BdbSrvUtil::_transMode       = oocNoOpen;
const int         BdbSrvUtil::_noReqPerLog     = 100;
unsigned long     BdbSrvUtil::_noCommits       = 0;
int               BdbSrvUtil::_lockWait        = 30;// sec
string         BdbSrvUtil::_lastErrorMsg    = "";
const int         BdbSrvUtil::MAX_NO_CACHED_DBS = 5;
BdbSrvUtil::DbName2Id* BdbSrvUtil::_dbs2ids     = new DbName2Id[MAX_NO_CACHED_DBS];
BdbMutex          BdbSrvUtil::_dbCacheMutex("dbs2idsMutex");

/**
 **  The function obtains an exclusive lock on a dummy database
 **  and holds that lock (in a separate context) as long as the 
 **  server is running.
 **/
BdbStatus
BdbSrvUtil::getExclusiveServerLock()
{
    const char* dbName = "clustServerLock";    

    _exclLockCtx = new ooContext(6,10); // small cache ok

    ooContext* mainContext = ooContext::current();
    ooContext::setCurrent(_exclLockCtx);
    cout << "BdbSrvUtil::getExclusiveServerLock set ooContext to _exclLockCtx = " << _exclLockCtx << endl;
    ooRegErrorHandler(BaBarObjyDummyErrorHandler);
    
    _exclLockTrans.start(oocNoMROW, 1); // 3 sec - lock wait time (should be enough)

    BdbHandle(BdbFDObj)   fdH;
    BdbHandle(BdbDBObj)   dbH;

    BdbStatus status = BdbcSuccess;
    if ( BdbcSuccess != fdH.open(fdName(), oocUpdate) ) {
        cerr << "Unable to open fd " << fdName() << endl;
        _exclLockTrans.abort();
        status = BdbcError;
        goto end;
    }
    
    if ( ! dbH.exist(fdH, dbName, oocNoOpen) ) {
        dbH = new (fdH) ooDBObj (dbName, 1, 1);
        if ( BdbIsNull(dbH) ) {
            cerr << "Unable to create db " << dbName << endl;
            _exclLockTrans.abort();
            status = BdbcError;
            goto end;
        }        
    } else {
        if ( BdbcSuccess != dbH.exist(fdH, dbName, oocUpdate) ) {
            cerr << "Unable to get exclusive server lock (e)" << endl;
            _exclLockTrans.abort();
            status = BdbcError;
            goto end;
        }
    }

    if ( BdbcSuccess != dbH.lock(oocLockUpdate) ) {
        cerr << "Unable to get exclusive server lock (l)" << endl;
        _exclLockTrans.abort();
        status = BdbcError;
        goto end;
    }

end:
    ooContext::setCurrent(mainContext);
    cout << "BdbSrvUtil::getExclusiveServerLock set ooContext to mainContext = " << mainContext << endl;
    if ( BdbcError == status ) {
        delete _exclLockCtx;
        _exclLockCtx = 0;
    }
    
    return status;
}


/**
 **   The function releases exclusive server lock
 **/
BdbStatus
BdbSrvUtil::releaseExclusiveServerLock()
{
    ooContext* mainContext = ooContext::current();
    ooContext::setCurrent(_exclLockCtx);
    cout << "BdbSrvUtil::releaseExclusiveServerLock set ooContext to _exclLockCtx = " << _exclLockCtx << endl;
    BdbStatus status = _exclLockTrans.commit();
    ooContext::setCurrent(mainContext);
    cout << "BdbSrvUtil::releaseExclusiveServerLock set ooContext to mainContext = " << mainContext << endl;
    if ( 0 != _exclLockCtx ) {
        delete _exclLockCtx;
        _exclLockCtx = 0;
    }
    
    return status;
}



/**
 **  The function opens the log file (if path != 0 )
 **  or it binds the log stream to the bdbdebug
 **/ 
BdbStatus
BdbSrvUtil::initLogging(const char* logDir)
{
    if ( 0 == logDir ) {
        cerr << "logDir = 0" << endl;
        return BdbcError;
    }

    if ( 0 != _logDir ) {
        cerr << "_logDir != 0" << endl;
        return BdbcError;
    }
    
    _logDir = new char [strlen(logDir)+1];
    strcpy(_logDir, logDir);

    // check if directory exists
    if ( 0 != access(_logDir, X_OK) ) {
        cerr << "Directory does not exist or wrong access rights" << endl;
        return BdbcError;
    }
    
    if ( BdbcSuccess != BdbSrvLogMgr::openLogFiles() ) {
        return BdbcError;
    }

    SRVLOG << PRINT_REQID << "Initialization" << endl;
    
    return BdbcSuccess;
}


BdbStatus
BdbSrvUtil::shutdown()
{
    if ( 0 != _logDir ) {
        delete [] _logDir;
        _logDir = 0;
    }
    if ( 0 != _chsMgr ) {
        delete _chsMgr;
        _chsMgr = 0;
    }
    if ( 0 != _dbmMgr ) {
        delete _dbmMgr;
        _dbmMgr = 0;
    }
    if ( 0 != _fsSelector ) {
        delete _fsSelector;
        _fsSelector = 0;
    }

    delete [] _dbs2ids;
    
    return BdbcSuccess;
}




BdbActiveDbSpec*
BdbSrvUtil::initSpec(const BdbGroupDescr& descr)
{
    BdbActiveDbSpec* spec = 0;
    if ( 0 == _chsMgr ) {
        SRVLOG << "_chsMgr not init" << endl;
        return spec;
    }
    
    const char* authLevelName = "User";
    if ( descr.authLevel() == 'g' ) {
        static const char* n = "Group";
        authLevelName = n;
    } else if ( descr.authLevel() == 's' ) {
        static const char* n = "System";
        authLevelName = n;
    }

    // retrieve all the info from transient cache maintained by BdbClustHintSetupMgr
    const  char* fullDName = BdbDomain::domainName(descr.sdName());

    // During parallel recovery many threads may call that at the same time.
    // Make sure the access is serialized.
    _chsMutex.lock();

    _chsMgr->setStrings(descr.clusterName(), fullDName, 
                        authLevelName, descr.authName(), descr.compName());
    
    d_ULong values[10];    
    _chsMgr->retrieveDomainSpec((unsigned*)values, 10);

    SRVLOG << "values for : " << fullDName << " "
           << descr.authLevel() << " "  << descr.authName()
           << " " << descr.compName() << "\n"
           << "   dirRange       = " << values[0] << "\n"
           << "   initNrPages    = " << values[1] << "\n"
           << "   percGrowth     = " << values[2] << "\n"
           << "   contHashing    = " << values[3] << "\n"
           << "   minNoPrecrDbs  = " << values[4] << "\n"
           << "   maxNoPrecrDbs  = " << values[5] << "\n"
           << "   precrNoConts   = " << values[6] << "\n"
           << "   maxDbSize      = " << values[7] << "\n"
           << "   prefContSize   = " << values[8] << "\n"
           << "   hardContLimit  = " << values[9] << endl;

    // convert all these values to what we really need
    // and use defaults if not defined

    spec = new BdbActiveDbSpec();

    int pageSize = BdbApplicationOrDomain::activeInstance()->fd().pageSize()/1024; // in KB
    spec->_hardPageLimit      = values[7] / pageSize;
    spec->_prefNoPagesPerCont = values[8] / pageSize;
    spec->_hardContLimit      = values[9];
    spec->_dirRange           = values[0];
    spec->_initNoPages        = values[1];
    spec->_percGrowth         = values[2];
    spec->_hashed             = values[3];
    spec->_minNoPrecrDbs      = values[4];
    spec->_maxNoPrecrDbs      = values[5];

    if ( values[7] < values[8] ) {
        SRVLOG << "WARNING: Incorrect values loaded via BdbFileConfigLoader: "
               << "(maxDbSize < prefContSize)" << endl;
        spec->_prefNoPagesPerCont = spec->_hardPageLimit;
    }
    spec->adjust();

    const char* cGroupName = _chsMgr->retrieveFSGroupName();
    char* groupName = 0;
    if ( ! _fsSelector->containsGroup(cGroupName) ) {
        vector<BdbTFileSystem*>* vector = 0;
        if ( BdbcSuccess != _chsMgr->retrieveFileSystems(groupName, vector) ) {
            SRVLOG << "ERR failed to retrieve file system info" << endl;
            delete spec;
            spec = 0;
            _chsMutex.unlock();
            return spec;
        }
        if ( BdbcSuccess != _fsSelector->addGroup(groupName, vector) ) {
            SRVLOG << "ERR failed to add group" << endl;
            delete spec;
            spec = 0;
            delete [] groupName;
            _chsMutex.unlock();
            return spec;
        }
    } else {
        groupName = new char [strlen(cGroupName)+1];
        strcpy(groupName, cGroupName);
    }
    
    spec->_fsGroupName = groupName;
    SRVLOG << "   fsGroupName    = " << groupName << endl;
    
    _chsMutex.unlock();
    return spec;
}



BdbStatus
BdbSrvUtil::buildClientKey(const char* hostName, int pid,
                                string& rwKey)
{
    if ( hostName == 0 ) {
        rwKey = "in-process";
        return BdbcSuccess;
    }
    
    int length = strlen(hostName)+10;
    char* s = new char [length];
    ostrstream stream(s, length);
    stream << hostName << pid << ends;

    rwKey = s;
    delete [] s;

    return BdbcSuccess;
}



char*
BdbSrvUtil::buildDbName(const char* sdName, 
                        const char* clusterName)
{
    const char* fn = "BdbSrvUtil::buildDbName";

    if ( 0 == sdName ) {
        SRVLOG << fn << "sdName == 0" << endl;
        return 0;
    }

    const char* n = "_clustHintConfigSrv"; 
    int len = strlen(sdName) + strlen(n) + 3;
    if ( clusterName != 0 ) {
        len += strlen(clusterName);
    }
    
    char* dbName = new char [len];
    
    strcpy(dbName, sdName);
    strcat(dbName, n);
    if ( 0 != clusterName ) {
        strcat(dbName, "_");
        strcat(dbName, clusterName);
    }
    SRVLOG << "Db name is " << dbName << endl;
    return dbName;
}



/**
 **  The function scans the catalog, finds all
 **  clustering hint databases, and locates all
 **  registered groups. It fills the set, the set
 **  contains BdbClSrvCRef*
 **  Transaction is required, can be run in main thread only
 **  (used app->fd())
 **/
BdbStatus
BdbSrvUtil::findAllGroups(set<BdbClSrvCRef*,babar::Collection::PtrLess>& theset)
{
    const char* fn = "BdbSrvUtil::findAllGroups";
    
    ooItr(ooDBObj) dbItr;
    if ( BdbcSuccess != BdbApplicationOrDomain::activeInstance()->fd().contains(dbItr) ) {
        SRVLOG << fn << " ERR Cannot initialize fd iterator" << endl;
        return BdbcError;
    }

    while ( dbItr.next() ) {
        const char* dbName = dbItr.name();
        if ( 0 != strstr(dbName, "clustHintConfigSrv") ) {
            // iterate over all containers
            ooRef(ooDBObj) dbR = dbItr;
            SRVLOG << "Iterating over all containers in db: " << dbItr.name() 
                   << ", oid = " << BdbPrintOID(dbR) << endl;
            ooItr(ooContObj) contItr;
            if ( dbR.contains(contItr, oocRead) != BdbcSuccess ) {
                SRVLOG << "Cannot initialize container iterator in db " << dbName << endl;
                return BdbcError;
            }
            // find all registered groups (one cont per group)
            while ( contItr.next() ) {
                const char* contName = contItr.name();

                ooRef(ooContObj) c = contItr;
                SRVLOG << "   found container: oid = " << BdbPrintOID(c)
                       << ", name = \"" << contName << "\"" << endl;
                
                if ( 0 != strcmp(contName, "_ooDefaultContObj") ) {
                    ooRef(ooContObj) contR = contItr;
                    theset.insert(new BdbClSrvCRef(contR));
                }
            }
        }
    }
    
    return BdbcSuccess;
}


/**
 **  Builds part of the full db path
 **  if event store: subDir     "evs/groups/isPhysicsEvents/aod/aod"
 **                  partDbName "evs_g_isPhysicsEvents_aod"
 **/
BdbStatus
BdbSrvUtil::constructPartDirAndDbName(const char* groupName, 
                                      char*& subDirs, 
                                      char*& partDbName)
{
    BdbGroupDescr descr;
    if ( BdbcSuccess != splitName(groupName, descr) ) {
        SRVLOG << "splitName falied for groupName = " << groupName << endl;
        return BdbcError;
    }
    
    if ( strcmp(descr.sdName(), "evs") ) {
        SRVLOG << "constructPartDirAndDbName not implemented for domain: " 
               << descr.sdName() << endl;
        return BdbcError;
    }
    
    const char* fullDName = BdbDomain::domainName(descr.sdName());
    
    const char* authLevel = BdbDomain::authLevelName(descr.authLevel());
    int length = strlen(fullDName) + strlen(authLevel) + 
                 strlen(descr.authName())   + 2 * strlen(descr.compName()) + 8;
    
    subDirs = new char[length];
    ostrstream pStream(subDirs, length);
    pStream << fullDName << "/" << authLevel << "/" << descr.authName()
            << "/" << descr.compName() << "/" << descr.compName() << ends;
    
    SRVLOG << "Relative part of directory: " << subDirs << endl;

    length = strlen(descr.authName()) + 16;
    partDbName = new char [length];
    ostrstream dStream(partDbName, length);
    dStream << descr.sdName() << "_" << descr.authLevel() << "_" 
            << descr.authName() << "_" << descr.compName() << ends;
    SRVLOG << "Part of db name: " << partDbName << endl;

    return BdbcSuccess;
}



/**
 **  The function constructs database path and database name, eg
 ** subDir and partDbName
 **  if event store: "evs/groups/isPhysicsEvents/aod/aod" "evs_g_isPhysicsEvents_aod"
 **/
BdbStatus
BdbSrvUtil::constructFinalNamesAndPath(const char* baseDir,
                                       const char* subDirs,
                                       const char* partDbName,
                                       char*& fullPath,
                                       char*& dbSystemName,
                                       char*& dbFileName,
                                       d_ULong dbid, 
                                       d_ULong dirRange)
{
    // build full path
    if ( baseDir == 0 || subDirs == 0 || partDbName == 0 ) {
        return BdbcError;
    }
    int length = strlen(baseDir) + strlen(subDirs) + 16;
    fullPath = new char [length];

    int leftRange  = ( dbid / dirRange ) * dirRange;
    int rightRange = leftRange + dirRange;
#define HEXA_FORMAT setfill('0') << setw(6) << hex << setiosflags(ios::uppercase) 
    ostrstream pStream(fullPath, length);
    pStream << baseDir << "/" << subDirs 
            << HEXA_FORMAT << leftRange << "-" 
            << HEXA_FORMAT << rightRange << ends;
    SRVLOG << "Full path: " << fullPath << endl;
    
    // build db system name
    length = strlen(partDbName) + 8;
    dbSystemName = new char [length];
    ostrstream dName(dbSystemName, length);
    dName << partDbName << HEXA_FORMAT << dbid << ends;
    SRVLOG << "Db system name: " << dbSystemName << endl;
#undef HEXA_FORMAT
    // build db file name
    length = strlen(dbSystemName) + 5;
    dbFileName = new char [length];
    ostrstream fName(dbFileName, length);
    fName << dbSystemName << ".bdb" << ends;
    SRVLOG << "Db file name: " << dbFileName << endl;

    return BdbcSuccess;
}

BdbStatus
BdbSrvUtil::startTrans(ooMode mode)
{
    if ( _transMode != oocNoOpen ) {
        SRVLOG << "ERR: trans already running" << endl;
        return BdbcError;
    }
    
    BdbStatus status;
    
    if ( mode == oocRead ) {
        status = BdbApplicationOrDomain::activeInstance()->startRead("ad");
    } else if ( mode == oocUpdate ) {
        status = BdbApplicationOrDomain::activeInstance()->startUpdate("ad");
    }
    
    if ( BdbcSuccess == status ) {
        _transMode = mode;
    }
    return status;
}

    
BdbStatus
BdbSrvUtil::commitTrans()
{
    if ( _transMode == oocNoOpen ) {
        SRVLOG << "ERR: trans not opened" << endl;
        return BdbcError;
    }
    
    BdbStatus status = BdbApplicationOrDomain::activeInstance()->commit("ad");
    if ( BdbcSuccess != status ) {
        SRVLOG << "ERR: cannot commit transaction" << endl;
    } else {
        _transMode = oocNoOpen;
    }
    _noCommits++;
    
    return status;
}


BdbStatus
BdbSrvUtil::abortTrans()
{
    if ( _transMode == oocNoOpen ) {
        SRVLOG << "ERR: trans not opened" << endl;
        return BdbcError;
    }
    
    BdbStatus status = BdbApplicationOrDomain::activeInstance()->commit("ad");
    if ( BdbcSuccess != status ) {
        SRVLOG << "ERR: cannot abort transaction" << endl;
    } else {
        SRVLOG << "Transaction aborted" << endl;
        _transMode = oocNoOpen;
    }
    return status;
}



unsigned long 
BdbSrvUtil::getReqID()
{
    return _reqID;
}


void 
BdbSrvUtil::incrReqID()
{
    _reqID++;
    _lastReqTime = time(0);
}




//raw_cl1_evs_g_isPhysicsEvents_stream
BdbStatus
BdbSrvUtil::splitName(const char* name, BdbGroupDescr& descr)
{
    SRVLOG << "BdbSrvUtil::splitName: name = " << name << endl;
    
    if ( 0 == name ) {
        return BdbcError;
    }
    
    int pos = 0;
    // comp name
    const char* p1 = strchr(name, '_');
    if ( p1 <= 0  ) {
        SRVLOG << "Cannot find first _ in " << name << endl;
        return BdbcError;
    }
    int len = p1 - name;
    char* compName = new char [len+1];
    strncpy(compName, name, len);
    compName[len] = '\0';
    pos += len+1;
    // cluster name
    const char* p2 = strchr(name+pos, '_');
    len = p2 - p1 - 1;
    char* clusterName = 0;
    if ( len > 0 ) {
        clusterName = new char [len+1];
        strncpy(clusterName, name+pos, len);
        clusterName[len] = '\0';
    }
    pos += len + 1;
    // domain name
    const char* p3 =  strchr(name+pos, '_');
    len = p3 - p2 - 1;
    if ( len <= 0 ) {
        SRVLOG << "Cannot find third _ in " << name << endl;
        return BdbcError;
    }
    char* sdName = new char [len+1];
    strncpy(sdName, name+pos, len);
    sdName[len] = '\0';
    pos += len+1;
    // auth level
    p3 += 2;
    char authLevel = *(name+pos);
    pos += 2;
    // auth name
    const char* p4 =  strchr(name+pos, '_');
    len = p4 - p3 - 1;
    if ( len <= 0 ) {
        SRVLOG << "Cannot find 4th _ in " << name << endl;
        return BdbcError;
    }
    char* authName = new char [len+1];
    strncpy(authName, name+pos, len);
    authName[len] = '\0';
    pos += len+1;
    // stream name
    char* streamName = 0;
    if ( name+pos > 0 ) {
        len = strlen(name+pos);
        streamName = new char [len+1];
        strcpy(streamName, name+pos);
    }
    if ( BdbcSuccess != descr.set(clusterName, sdName, authLevel, 
                                  authName, compName, streamName) ) {
        return BdbcError;
    }
    if ( 0 != clusterName ) delete [] clusterName;
    if ( 0 != sdName      ) delete [] sdName;
    if ( 0 != authName    ) delete [] authName;
    if ( 0 != compName    ) delete [] compName;
    if ( 0 != streamName  ) delete [] streamName;
        
    return BdbcSuccess;
}


/**
 **  The function converts group name to short domain name.
 **  Function calling this one should deallocate the string
 **/
char* 
BdbSrvUtil::groupName2SDName(const char* groupName)
{
    BdbGroupDescr descr;
    if ( BdbcSuccess != splitName(groupName, descr) ) {
        return 0;
    }
    
    char* s = new char [strlen(descr.sdName())+1];
    strcpy(s, descr.sdName());
    return s;
}




    

const char* 
BdbSrvUtil::fdName()
{
    return BdbApplicationOrDomain::activeInstance()->bootName();
}




void
BdbSrvUtil::cacheDbInfo(const char* name, int id)
{
    _dbCacheMutex.lock();
    
    int i;
    d_Boolean  full = d_True;
    for (i=0 ; i<MAX_NO_CACHED_DBS ; i++) {
        if ( (_dbs2ids[i]).id == id ) {
            _dbCacheMutex.unlock();
            cout << "entry already in (id = " << id << ")" << endl;
            return;
        }
        if ( (_dbs2ids[i]).id == 0 ) {
            full = d_False;
        }
        
    }
    if ( full ) {
        _dbCacheMutex.unlock();
        cout << "Cannot cache more than " << MAX_NO_CACHED_DBS << " (cache full)" << endl;
        return;
    }
    
    for (i=0 ; i<MAX_NO_CACHED_DBS ; i++) {
        if ( (_dbs2ids[i]).id == 0 ) {
            (_dbs2ids[i]).set(name, id);
            _dbCacheMutex.unlock();
            cout << "entry cached on pos " << i << endl;
            return;
        }
    }
    _dbCacheMutex.unlock();
}

int
BdbSrvUtil::getDbId(const char* dbName)
{
    int i, id = 0;
    _dbCacheMutex.lock();
    for (i=0 ; i<MAX_NO_CACHED_DBS ; i++) {
        if ( (_dbs2ids[i]).itIsMe(dbName) ) {
            id = (_dbs2ids[i]).id;
            break;
        }
    }
    _dbCacheMutex.unlock();
    return id;
}

    

 

---

BaBar Public Site | SLAC | News | Links | Who's Who | Contact Us

Page Owner: Jacek Becla
Last Update: October 04, 2002