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

                 

---

/BdbModules/BdbTransactionMgr.cc

Go to the documentation of this file.

//--------------------------------------------------------------------------
// File and Version Information:
//      $Id: BdbTransactionMgr.cc,v 1.22 2002/06/21 19:13:37 ryd Exp $
//
// Description:
//      Class BdbTransactionMgr implementation file.
//
// Environment:
//      Software developed for the BaBar Detector at the SLAC B-Factory.
//
// Author List:
//      David R. Quarrie        Original Author
//
// Copyright Information:
//      Copyright (C) 1999      Lawrence Berkeley National Laboratory
//
//------------------------------------------------------------------------
#include "BaBar/BaBar.hh"

//-----------------------
// This Class's Header --
//-----------------------
#include "BdbModules/BdbTransactionMgr.hh"

//-------------
// C Headers --
//-------------
extern "C" {
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#include <sys/time.h>
}

//---------------
// C++ Headers --
//---------------

//-------------------------------
// Collaborating Class Headers --
//-------------------------------
#include "AbsEvent/AbsEvent.hh"
#include "BdbApplication/BdbDebug.hh"
#include "BdbEventStore/BdbEventStore.hh"
#include "BdbCond/BdbConditions.hh"
#include "BdbConfig/BdbConfig.hh"
#include "BdbModules/BdbTransactionClient.hh"
#include "ErrLogger/ErrLog.hh"
#include <string>
using std::string;

//-----------------------------------------------------------------------
// Local Macros, Typedefs, Structures, Unions and Forward Declarations --
//-----------------------------------------------------------------------

static const char rscid[] = "$Id: BdbTransactionMgr.cc,v 1.22 2002/06/21 19:13:37 ryd Exp $";

#include "BdbApplication/BdbCheckPointFunc.hh"

/**
 * BdbForceCommitAndStart is a decorator class which forces commit/start
 * checkpoint behaviour irrespective of the setting of 'transNotOpen'
 * of the original function.
 */
class BdbForceCommitAndStart
   : public BdbCheckPointFunc
{
   public:
      BdbForceCommitAndStart( BdbCheckPointFunc& func ) ;
      virtual ~BdbForceCommitAndStart() ;

      // Selectors (const)
      virtual bool transNotOpen() ;

      // Modifiers
      virtual bool execute() ;

   private:
      BdbForceCommitAndStart() ; // stop default
      BdbForceCommitAndStart( BdbForceCommitAndStart& lhs ) ; // stop default
      BdbForceCommitAndStart& operator=( BdbForceCommitAndStart& lhs ) ; // stop default

      // data
      BdbCheckPointFunc& _func ;

} ;

BdbForceCommitAndStart::BdbForceCommitAndStart( BdbCheckPointFunc& func )
   : _func( func )
{
}

BdbForceCommitAndStart::~BdbForceCommitAndStart()
{
}

bool
BdbForceCommitAndStart::transNotOpen()
{
   return( !false ) ;
}

bool
BdbForceCommitAndStart::execute()
{
   _func.executeFunc() ;
   return( !false ) ;
}

static const string theName( "BdbTransactionMgr" );

//      ----------------------------------------
//      -- Static Function Member Definitions --
//      ----------------------------------------

const string&
BdbTransactionMgr::transactionMgrName()
{
   return( theName ) ;
}

//              ----------------------------------------
//              -- Public Function Member Definitions --
//              ----------------------------------------

//----------------
// Constructors --
//----------------

BdbTransactionMgr::BdbTransactionMgr( const char* const theName, 
                              const char* const theDescription )
  : AppModule( theName, theDescription ), 
     _granularity          ( "granularity", this, 60 ),
     _granEvents           ( "granEvents", this, false ),
     _granRandomize        ( "granRandomize", this, true ) ,
     _rpcTimeout           ( "rpcTimeout", this, 100 ),
     _cacheInit            ( "cacheInitPages", this, 0 ) ,
     _cacheMax             ( "cacheMaxPages", this, 0 ) ,
     _fileDescriptors      ( "fileDescriptors", this, 0 ) ,
     _checkVTablePointer   ( "checkVTablePointer", this, false ),
     _statistics           ( "statistics", this, false ),
     _firstEventTransaction( "firstEventTransaction", this, true ),
     _commitAndHoldLimit   ( "commitAndHoldLimit", this, 0 ),
     _openInitialConTrans  ( "openInitialConTrans", this, false ),
     _openInitialCfgTrans  ( "openInitialCfgTrans", this, false ),
     _handlePrintFreq      ( "handlePrintFrequency", this, 0 ),
     _handlePrintLevel     ( "handlePrintLevel", this, 0 ),
     _handlePrintLimit     ( "handlePrintLimit", this, 0 ),
     _isPrepared           ( false ),
     _useGranularity       ( 0 ),
     _totalCount           ( 0 ),
     _granularityCount     ( 0 ),
     _hasCommitted         ( false ),
     _loggingInput         ( false ),
     _loggingOutput        ( false ),
     _lastTime             ( time( (time_t*)0 ) ),
     _commitAndHoldCount   ( 0 ),
     _openedInitialConTrans( false ),
     _openedInitialCfgTrans( false ),
     _handlePrintCount     ( 0 )
{
    // Add parameters to list of command handlers
    commands( )->append( &_statistics );
    commands( )->append( &_granularity );
    commands( )->append( &_granEvents );
    commands( )->append( &_granRandomize );
    commands( )->append( &_rpcTimeout );
    commands( )->append( &_cacheInit );
    commands( )->append( &_cacheMax );
    commands( )->append( &_fileDescriptors );
    commands( )->append( &_checkVTablePointer );
    commands( )->append( &_firstEventTransaction );
    commands( )->append( &_commitAndHoldLimit );
    commands( )->append( &_openInitialConTrans );
    commands( )->append( &_openInitialCfgTrans );
    commands( )->append( &_handlePrintFreq );
    commands( )->append( &_handlePrintLevel );
    commands( )->append( &_handlePrintLimit );
}

//--------------
// Destructor --
//--------------

BdbTransactionMgr::~BdbTransactionMgr( )
{
}

//-------------
// Methods   --
//-------------

AppResult
BdbTransactionMgr::beginJob( AbsEvent* anEvent )
{
    if ( _verbose.value( ) ) {
      ErrMsg( trace ) << name()
                      << ": begin Job"
                      << endmsg;
    }
    return AppResult::OK ;
}

AppResult
BdbTransactionMgr::beginRun( AbsEvent* anEvent )
{
    if ( _verbose.value( ) ) {
      ErrMsg( trace ) << name()
                      << ": begin Run"
                      << endmsg;
    }
    return AppResult::OK ;
}

AppResult
BdbTransactionMgr::event( AbsEvent* aEvent )
{

   return( AppResult::OK ) ;
}

AppResult
BdbTransactionMgr::endJob( AbsEvent* anEvent )
{
    if ( _verbose.value( ) ) {
        ErrMsg( trace ) << name()
                      << ": end Job"
                      << endmsg;
    }
    return AppResult::OK ;
}

void
BdbTransactionMgr::help( int argc, char** argv )
{
    cout << endl;

    cout << "This module is " << name( ) << " and was thought to be able" << endl;
    cout << "to " << description( ) << " when it was added" << endl;
    cout << "to this executable." << endl << endl;

    cout << "In reality it is an input module for the BaBar Event Store." << endl;
    cout << "This version currently works only for the EmcGHit data &" << endl;
    cout << "assumes that that data is initially transient. It is therefore" << endl;
    cout << "a testbed for future development rather than an end product." << endl << endl;

    cout << "This module has one parameter, collectionName. This" << endl;
    cout << "specifies the collection name (for the current user) that will be" << endl;
    cout << "read." << endl;
    cout << endl;

    AppModule::help( argc, argv );
}

void
BdbTransactionMgr::conditionalStart( BdbTransactionClient& client )
{
   if( client.hasCommitted() ) {
      return ;
   }

   // open initial Conditions transaction is necessary
   BdbConditions& conditions( *BdbConditions::instance() ) ;
   if( ! conditions.transIsOpen() &&
       client.openInitialConTrans() ) {
      conditions.change( BdbcRead ,
                         d_True ) ;
      client.setOpenedInitialConTrans() ;
   }

   // open initial Configuration transaction is necessary
   BdbConfig& config( *BdbConfig::instance() ) ;
   if( ! config.transIsOpen() &&
       client.openInitialCfgTrans() ) {
      config.change( BdbcRead ,
                     d_True ) ;
      client.setOpenedInitialCfgTrans() ;
   }
}

bool
BdbTransactionMgr::conditionalCheckPoint( BdbTransactionClient& client ,
                                          BdbCheckPointFunc& func )
{
    BdbEventStore* theStore = BdbEventStore::instance( );
    bool result = false;

    // Output handle information is requested by the client
    if( 0 != client.handlePrintFrequency() ) {
       client.incrementHandlePrintCount() ;
       if( client.handlePrintFrequency() == client.handlePrintCount() ) {
          opiPrintHList( client.handlePrintLevel() ,
                         client.handlePrintLimit() ) ;
          client.resetHandlePrintCount() ;
       }
    }

    // Commit the transaction if the granularity threshold has been
    // exceeded. This threshold is determined by the granularity and 
    // granEvents parameters. The granEvents parameter determines
    // whether the units are events (true) or seconds (false). The
    // granularity parameter specifies the number of units (events or
    // seconds) for the threshold.

    if( ( client.granularityIsEvents() &&
          ( client.eventCount() >= client.granularity() ) ) ||
        ( ! client.granularityIsEvents() &&
          ( ((size_t) difftime( time( (time_t*)0 ), client.lastTime() )) >= client.granularity() ) ) ) {

        // Transaction granularity exceeded - commit this transaction
        if ( client.verboseOutput() ) {
            ErrMsg( routine ) << client.clientName()
                              << ": about to commit at "
                              << BdbTimeStamp( )
                              << endmsg;
        }

        BdbStatus checkPointResult( BdbcError ) ;
        if( 0 == client.commitAndHoldLimit() ) {
           BdbForceCommitAndStart commitAndStart( func ) ;
           checkPointResult = theStore->checkPoint( commitAndStart ) ;
        }
        else {
           if( client.commitAndHoldCount() >= client.commitAndHoldLimit() ) {
              BdbForceCommitAndStart commitAndStart( func ) ;
              checkPointResult = theStore->checkPoint( commitAndStart ) ;
              client.resetCommitAndHoldCount() ;
           }
           else {
              checkPointResult = theStore->checkPoint( func ) ;
              client.incrementCommitAndHoldCount() ;
           }
        }
        if ( BdbcSuccess != checkPointResult ) {
            ErrMsg( fatal ) << client.clientName()
                            << ": transaction checkpoint failed at "
                              << BdbTimeStamp( )
                              << " - Event: "
                            << client.totalEventCount()
                              << endmsg;
            ::abort( );
        }
        result = true;
        if ( client.verboseOutput() ) {
            ErrMsg( routine ) << client.clientName()
                              << ": transaction committed at "
                              << BdbTimeStamp( )
                              << " - Event: "
                              << client.totalEventCount()
                              << endmsg;
            ErrMsg( routine ) << client.clientName()
                              << ": new transaction started at "
                              << BdbTimeStamp( )
                              << endmsg;
        }

        if( ! client.hasCommitted() ) {

            // close initial Conditions transaction is necessary
            BdbConditions& conditions( *BdbConditions::instance() ) ;
            if( conditions.transIsOpen() &&
                client.openedInitialConTrans() ) {
                conditions.change( BdbcNoOpen ,
                                   d_True ) ;
            }

            // close initial Configuration transaction is necessary
            BdbConfig& config( *BdbConfig::instance() ) ;
            if( config.transIsOpen() &&
                client.openedInitialCfgTrans() ) {
                config.change( BdbcNoOpen ,
                               d_True ) ;
            }
        }

        // Reset counters & timers (and set 'hasCommitted')
        client.resetEventCount() ;
        client.resetLastTime() ;
    }
    return result;
}

bool
BdbTransactionMgr::conditionalCheckPoint( APPInputModule& theModule ,
                                          BdbTransactionClient& client ,
                                          BdbCheckPointFunc& func )
{
    client.incrementEventCount() ;
    client.inputLogged() ;

    bool result = false;
    // if first event has not been processed assume output is being logged.
    if ( ( ! ( client.loggingOutput() ) ) &&
         ( ( 0 != client.totalEventCount() ) &&
           ( 1 != client.totalEventCount() ) ) ) {
        result = conditionalCheckPoint( client ,
                                        func );
    }
    return result;
}

bool
BdbTransactionMgr::conditionalCheckPoint( APPOutputModule& theModule ,
                                          BdbTransactionClient& client ,
                                          BdbCheckPointFunc& func )
{
    if( ! client.loggingInput() ) {
       client.incrementEventCount() ;
    }
    client.outputLogged() ;

    return( conditionalCheckPoint( client ,
                                   func ) ) ;
}

//              -------------------------------------------
//              -- Protected Function Member Definitions --
//              -------------------------------------------

void
BdbTransactionMgr::prepareObjy( BdbTransactionClient& client )
{
   BdbEventStore*  theStore = BdbEventStore::instance( );
   if ( ! theStore->oodbIsInitialized( ) ) {

      // Objectivity hasn't yet been initialized - do it now
      // Setup the Objectivity initial cache size (in pages). A value of 0
      // leaves the Objectivity default unchanged.
      d_ULong newValue = client.cacheInit( );
      if ( 0 < newValue ) {
         theStore->setCacheInitPages( newValue );
      }
      ErrMsg( routine ) << client.clientName( )
                        << ": Objectivity initial cache size set to "
                        << theStore->cacheInitPages( )
                        << " pages." << endmsg;

      // Setup the Objectivity maximum cache size (in pages). A value of 0
      // leaves the Objectivity default unchanged.
      newValue = client.cacheMax( );
      if ( 0 < newValue ) {
         theStore->setCacheMaxPages( newValue );
      }
      ErrMsg( routine ) << client.clientName( )
                        << ": Objectivity maximum cache size set to "
                        << theStore->cacheMaxPages( )
                        << " pages." << endmsg;

      // Setup the Objectivity number of file descriptors. A value of 0
      // leaves the Objectivity default unchanged.
      newValue = client.fileDescriptors( );
      if ( 0 < newValue ) {
         theStore->setFileDescriptors( newValue );
      }
      ErrMsg( routine ) << client.clientName( )
                        << ": Objectivity number of file descriptors set to "
                        << theStore->fileDescriptors( )
                        << endmsg;

      // Setup the RPC Timeout interval. However, don't reduce it unless the
      // new value is zero, in which case a 60sec timeout is established by default.
      d_ULong newRpc = client.rpcTimeout( );
      d_ULong oldRpc = theStore->rpcTimeout( );
      if ( 0 == newRpc ) {
         theStore->setRpcTimeout( 60 );
      } else {
         if ( newRpc > oldRpc ) {
            theStore->setRpcTimeout( newRpc );
         }
      }
      ErrMsg( routine ) << client.clientName( )
                        << ": RPC Timeout set to "
                        << theStore->rpcTimeout( )
                        << "secs." << endmsg;

      // Setup whether to check the VTable Pointers
      theStore->setCheckVTablePointer( client.checkVTablePointers( ) );
      if( theStore->isVTablePointerChecked() ) {
         ErrMsg( routine ) << client.clientName( )
                           << ": VTable Pointers are checked."
                           << endmsg;
      }
      else {
         ErrMsg( routine ) << client.clientName( )
                           << ": VTable Pointers are not checked."
                           << endmsg;
      }
   }
   else {
      ErrMsg( routine ) << client.clientName( )
                        << ": Objectivity is already initialized - "
                        << "initialization bypassed."
                        << endmsg;
   }
}

void
BdbTransactionMgr::prepare( BdbTransactionClient& client )
{
    BdbEventStore*  theStore = BdbEventStore::instance( );
    if ( ! client.isPrepared() ) {

        prepareObjy( client ) ;


        // Ensure that data is readable only by users who belong to the
        // eventstore domain
        theStore->startNestedRead() ;
        if ( ! theStore->isAccessAllowed( theStore->userName( ) ) ) {
            ErrMsg( fatal ) << "**************************************************" << endl
                            << "User " << theStore->userName( ) 
                            << ", you are not a registered user of the "            << endl
                            << "event store for this federation. Access is denied." << endl
                            << "**************************************************" << endmsg;
            theStore->abort( );
            ::abort( );
        }
        theStore->commitNested() ;

        if ( client.granularityIsEvents( ) ) {
            ErrMsg( routine ) << client.clientName( )
                              << ": Transaction granularity set to "
                              << client.granularity()
                              << " events."
                              << endmsg;
        } else {
            ErrMsg( routine ) << client.clientName( )
                              << ": Transaction granularity set to "
                              << client.granularity()
                              << " secs."
                              << endmsg;
        }
        client.setPrepared() ;
    }
}

bool
BdbTransactionMgr::isPrepared() const
{
   return( _isPrepared ) ;
}

int
BdbTransactionMgr::cacheInit() const 
{
   return( _cacheInit.value() ) ;
}

int
BdbTransactionMgr::cacheMax() const 
{
   return( _cacheMax.value() ) ;
}

int
BdbTransactionMgr::fileDescriptors() const 
{
   return( _fileDescriptors.value() ) ;
}

int
BdbTransactionMgr::rpcTimeout() const 
{
   return( _rpcTimeout.value() ) ;
}

bool
BdbTransactionMgr::checkVTablePointers() const 
{
   return( _checkVTablePointer.value() ) ;
}


bool
BdbTransactionMgr::loggingInput() const
{
   return( _loggingInput ) ;
}

bool
BdbTransactionMgr::loggingOutput() const
{
   return( _loggingOutput ) ;
}

const char*
BdbTransactionMgr::clientName() const
{
   return( name() ) ;
}

bool
BdbTransactionMgr::verboseOutput() const
{
   return( _verbose.value() ) ;
}

bool
BdbTransactionMgr::granularityIsEvents() const
{
   return( _granEvents.value() ) ;
}

size_t
BdbTransactionMgr::granularity() const
{
   if( 0 != _useGranularity ) {
      return( _useGranularity ) ;
   }

   // Setup the effective granularity counter. This is the
   // "granularity" value, unless the "granRandomize" flag is set
   // to true, in which the the effective granularity is the sum of
   // 1/2 the requested granularity plus (milisec of current time % requested
   // granularity). This adds a random component (albeit forming
   // constant effective granularity for each job) in the case of
   // multiple jobs. Note also that a requested granularity of zero
   // or less restores the original default.
   size_t valueToUse( _granularity.value() ) ;
   if ( 0 == valueToUse ) {
      valueToUse = 60;
   }
   if ( _granRandomize.value( ) ) {
      struct timeval tv;
      gettimeofday(&tv, 0);
      int delta = tv.tv_usec % valueToUse;
      valueToUse = valueToUse / 2;
      valueToUse += delta;
   }
   if ( valueToUse < 1 ) {
      valueToUse = 1;
   }
   const_cast< BdbTransactionMgr* >( this )->_useGranularity = valueToUse ;

   return( _useGranularity ) ;
}

size_t
BdbTransactionMgr::eventCount() const
{
   return( _granularityCount ) ;
}

size_t
BdbTransactionMgr::totalEventCount() const
{
   return( _totalCount ) ;
}

time_t
BdbTransactionMgr::lastTime() const
{
   return( _lastTime ) ;
}

bool
BdbTransactionMgr::hasCommitted() const
{
   return( _hasCommitted ) ;
}

size_t
BdbTransactionMgr::commitAndHoldLimit() const
{
   return( _commitAndHoldLimit.value() ) ;
}

size_t
BdbTransactionMgr::commitAndHoldCount() const
{
   return( _commitAndHoldCount ) ;
}

bool
BdbTransactionMgr::openInitialConTrans() const
{
   return( _openInitialConTrans.value() ) ;
}

bool
BdbTransactionMgr::openedInitialConTrans() const
{
   return( _openedInitialConTrans ) ;
}

bool
BdbTransactionMgr::openInitialCfgTrans() const
{
   return( _openInitialCfgTrans.value() ) ;
}

bool
BdbTransactionMgr::openedInitialCfgTrans() const
{
   return( _openedInitialCfgTrans ) ;
}

size_t
BdbTransactionMgr::handlePrintFrequency() const 
{
   return( _handlePrintFreq.value() ) ;
}

size_t
BdbTransactionMgr::handlePrintCount() const 
{
   return( _handlePrintCount ) ;
}

size_t
BdbTransactionMgr::handlePrintLevel() const 
{
   return( _handlePrintLimit.value() ) ;
}

size_t
BdbTransactionMgr::handlePrintLimit() const 
{
   return( _handlePrintLimit.value() ) ;
}

void
BdbTransactionMgr::setPrepared()
{
   _isPrepared = true ;
}

void
BdbTransactionMgr::inputLogged()
{
   _loggingInput = true ;
}

void
BdbTransactionMgr::outputLogged()
{
   _loggingOutput = true ;
}

void
BdbTransactionMgr::incrementEventCount()
{
   _totalCount++ ;
   _granularityCount++ ;
}

void
BdbTransactionMgr::resetEventCount()
{
   _hasCommitted = true ;
   _granularityCount = 0 ;
}

void
BdbTransactionMgr::resetLastTime()
{
   _hasCommitted = true ;
   _lastTime =  time( (time_t*)0 ) ;
}

void
BdbTransactionMgr::incrementCommitAndHoldCount()
{
   _commitAndHoldCount++ ;
}

void
BdbTransactionMgr::resetCommitAndHoldCount()
{
   _commitAndHoldCount = 0 ;
}

void
BdbTransactionMgr::setOpenedInitialConTrans()
{
   _openedInitialConTrans = !false ;
}

void
BdbTransactionMgr::setOpenedInitialCfgTrans()
{
   _openedInitialCfgTrans = !false ;
}

void
BdbTransactionMgr::incrementHandlePrintCount()
{
   _handlePrintCount++ ;
}

void
BdbTransactionMgr::resetHandlePrintCount()
{
   _handlePrintCount = 0 ;
}

 

---

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

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