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

                 

---

/BdbModules/BdbEventOutput.cc

Go to the documentation of this file.

//--------------------------------------------------------------------------
// File and Version Information:
//      $Id: BdbEventOutput.cc,v 1.151.2.3 2002/07/25 23:02:37 becla Exp $
//
// Description:
//      Class BdbEventOutput implementation file.
//
// Environment:
//      Software developed for the BaBar Detector at the SLAC B-Factory.
//
// Author List:
//      David Quarrie           Lawrence Berkeley National Laboratory
//
// Copyright Information:
//      Copyright (C) 1999      Lawrence Berkeley National Laboratory
//
//------------------------------------------------------------------------

#include "BaBar/BaBar.hh"

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

//-------------
// C Headers --
//-------------
#include <stdlib.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#include <sys/time.h>

//-------------------------------
// Collaborating Class Headers --
//-------------------------------

#include <vector>
using std::vector;
#include <string>
using std::string;
#include <algorithm>

#include "ErrLogger/ErrLog.hh"
#include "ProxyDict/Ifd.hh"
#include "AbsEvent/AbsEventID.hh"
#include "AbsEnv/AbsEnv.hh"
#include "GenEnv/GenEnv.hh"
#include "AbsEvent/AbsEvent.hh"
#include "AbsEvent/AbsEvtObj.hh"
#include "HepODBMS/tagdb/HepGenericTag.h"
#include "BdbApplication/BdbDebug.hh"
#include "BdbTrees/BdbHeaderKeysMgr.hh"
#include "BdbEvent/BdbEventT.hh"
#include "BdbEvent/BdbStateIdT.hh"
#include "BdbConverters/BdbAbsConverter.hh"
#include "BdbScribes/BdbEvtObjLocation.hh"
#include "BdbScribes/BdbConversionManager.hh"
#include "BdbEventStore/BdbEventStore.hh"
#include "BdbEventStore/BdbOptNoEventStorage.hh"
#include "BdbEventStore/BdbEvsPlacementManager.hh"
#include "BdbEvent/BdbEventT.hh"
#include "BdbEvent/BdbEvent_001.hh"
#include "BdbEventStore/BdbTreeCollectionT.hh"
#include "BdbEventStore/BdbVectorCollectionP.hh"
#include "BdbEventStore/BdbCollectionP.hh"
#include "BdbModules/BdbCollectionCommand.hh"
#include "BdbModules/BdbOutputCommand.hh"
#include "BdbModules/BdbOutputStream.hh"
#include "BdbModules/BdbEventInput.hh"
#include "BdbModules/BdbPtrProxy.hh"
#include "BdbModules/BdbTransactionMgr.hh"
#include "BdbModules/BdbModulesUtility.hh"
#include "CLHEP/Alist/AList.h"
#include "CLHEP/Alist/AIterator.h"
#include "EidData/EidCondKeyTriplet.hh"
#include "Framework/APPFramework.hh"
#include "FrameUtil/APPList.hh"
#include "FrameUtil/APPListIterator.hh"

//------------------------------------
// Collaborating Class Declarations --
//------------------------------------
class BdbAbstractClusteringHint;

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

static const IfdStrKey eventKey( "BdbNewEvent" ) ;
static const IfdStrKey collectionKey( "OutputCollection" ) ;
static const IfdStrKey collectionProxy( "OutputCollectionProxy" ) ;

static const BdbRef( BdbHeaderKeysMgr )& getHeaderKeys( const BdbHandle( BdbCollectionP )& collection )
{
   // Create global instance of lookup table
   const size_t kMaximumNumberCollections( 256 ) ;
   static BdbRef( BdbCollectionP ) collections[ kMaximumNumberCollections ] ;
   static BdbRef( BdbHeaderKeysMgr ) headerKeysMgrs[ kMaximumNumberCollections ] ;

   size_t index( 0 ) ;
   BdbRef( BdbCollectionP ) collectionR = BdbRef( BdbCollectionP)( collection ) ;
   while( ( kMaximumNumberCollections != index ) &&
          ( ! (collections[ index ]).isNull() ) &&
          ( collections[ index ] != collectionR ) ) {
      ++index ;
   }

   if( collections[ index ] ==  collectionR ) {
      return( headerKeysMgrs[ index ] ) ;
   }

   if( kMaximumNumberCollections == index ) {
      ErrMsg( fatal ) << "BdbEventOutput"
                      << ": kNumberOfCollections exceeded, please increase."
                      << endmsg ;
      ::abort() ;
   }

   collections[ index ] = collectionR ;
   BdbHandle( BdbHeaderKeysMgr ) mgr ;
   mgr = new( collectionR )BdbHeaderKeysMgr ;
   headerKeysMgrs[ index ] = BdbRef( BdbHeaderKeysMgr )( mgr ) ;
   return( headerKeysMgrs[ index ] ) ;
}

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

void
BdbEventOutput::copyTagDescriptor( BdbAbsCollectionT<BdbEventT>* aCollection ,
                                   BdbEventT& aEvent )
{
   // Ensure that this collection has a reference to the
   // tag description object for later iteration. This
   // implementation copies the tag description object from
   // the owner collection to the current collection.

   // WARNING: This might have problems if the description
   // for the owner collection is gradually appended to,
   // since later additions will not be tracked.

   BdbHandle(BdbCollectionP) myColl = aCollection->persistent() ;
   BdbHandle(HepExplorableDescr) theDescr;
   myColl->description( theDescr );
   if ( BdbIsNull( theDescr ) ) {
      BdbHandle(BdbCollectionP) theOwner;
      aEvent.owner( theOwner );
      if ( ! BdbIsNull( theOwner ) ) {
         BdbHandle(HepExplorableDescr) theDescr;
         theOwner->description( theDescr );
         if ( ! BdbIsNull( theDescr ) ) {
            myColl->setDescription( theDescr );
         }
      }
   }
}

//----------
// Public --
//----------

const IfdStrKey&
BdbEventOutput::newBdbEventKey()
{
   return( eventKey ) ;
}

const IfdStrKey&
BdbEventOutput::outputCollectionKey()
{
   return( collectionKey ) ;
}

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

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

BdbEventOutput::BdbEventOutput( const char* const theName, 
                                const char* const theDescription )
   : AppStreamsOutputModule( theName, theDescription ), 
     _forceNewDatabases    ( "forceNewDatabases", this, false ),
     _allInOne             ( "allInOne", this, false ),
     _reclustering         ( "reclustering", this, false ),
     _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 ),
     _renewHdrList         ( "renewHdrList", this, "" ),
     _statistics           ( "statistics", this, false ),
     _skimEvent            ( "skimEvent", this, false ),
     _useStreamPlacement   ( "useStreamPlacement", this, false ),
     _firstEventTransaction( "firstEventTransaction", this, true ),
     _openInitialConTrans  ( "openInitialConTrans", this, false ),
     _openInitialCfgTrans  ( "openInitialCfgTrans", this, false ),
     _commitStatistics     ( "commitStatistics", this, false ),
     _tokenizedHdrList     ( 0 ) ,
     _isPrepared           ( false ),
     _hasCommitted         ( false ) ,
     _useGranularity       ( 0 ),
     _total                ( 0 ),
     _granularityCount     ( 0 ),
     _lastTime             ( time( (time_t*)0 ) ) ,
     _openedInitialConTrans( false ),
     _openedInitialCfgTrans( false ),
     _transactionMgr       ( 0 )
{
    // Add parameters to list of command handlers
    commands( )->append( &_forceNewDatabases );
    commands( )->append( &_allInOne );
    commands( )->append( &_reclustering );
    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( &_renewHdrList );
    commands( )->append( &_skimEvent );
    commands( )->append( &_useStreamPlacement );
    commands( )->append( &_firstEventTransaction );
    commands( )->append( &_openInitialConTrans );
    commands( )->append( &_openInitialCfgTrans );
    commands( )->append( &_commitStatistics );
    commands( )->append( _collectionCmd = new BdbCollectionCommand( "collection"    , this ) );
    commands( )->append( _collCmd       = new BdbCollectionCommand( "coll"          , this ) );
    commands( )->append( _outputCmd     = new BdbOutputCommand    ( "output"        , this ) );
}

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

BdbEventOutput::~BdbEventOutput()
{
    delete _collectionCmd;
    delete _collCmd;
    delete _tokenizedHdrList;
}

//-----------------------------------------------
// Framework Module Interface Member Functions --
//-----------------------------------------------

AppResult
BdbEventOutput::beginJob( AbsEvent* anEvent )
{
    if ( _verbose.value( ) ) {
        ErrMsg( routine ) << name()
                        << ": begin Job at "
                        << BdbTimeStamp( )
                        << endmsg;
    }

    // Perform initial preparations & start the first transaction
    initialize( );

    return AppResult::OK ;
}


AppResult
BdbEventOutput::beginRun( AbsEvent* anEvent )
{
    if ( _verbose.value( ) ) {
        ErrMsg( routine ) << name()
                        << ": begin Run at "
                        << BdbTimeStamp( )
                        << endmsg;
    }

    BdbEventStore* theStore = BdbEventStore::instance( );
    theStore->setLockWait( 300 );
    theStore->activate() ;

    // Start an update transaction if not already in one
    BdbMode oldMode = theStore->change( BdbcUpdate );
    AppResult result = AppStreamsOutputModule::beginRun( anEvent );

    // Restore the transaction state and force a transaction commit
    theStore->change( oldMode, d_True );
    theStore->deactivate( );

    conditionalStart() ;

    return result;
}

AppResult
BdbEventOutput::outputEvent( AbsEvent* anEvent )
{
    // Perform output for the current event
    bool status = doEvent( anEvent );

    // Commit the transaction if the granularity threshold has been
    // exceeded. A new transaction is also started if necessary. At 
    // the same time we stall if the federation is marked as being 
    // unavailable.
    bool committed = conditionalCommit( );

    return AppResult::OK;
}

AppResult
BdbEventOutput::endJob( AbsEvent* anEvent )
{
    BdbStatus status;

    if ( _verbose.value( ) ) {
        ErrMsg( routine ) << name( )
                        << ": end Job at "
                        << BdbTimeStamp( )
                        << endmsg;
    }

    // Start an update transaction if not already in one
    BdbEventStore* theStore = BdbEventStore::instance( );
    theStore->activate() ;
    if( ! theStore->isStandalone() ) {
       theStore->change( BdbcUpdate );
    }

    AppResult result( AppStreamsOutputModule::endJob( anEvent ) );

    // Commit the transaction if still open
    doCommit();

    // Loop over output streams printing their statistics information
    BdbOutputStream** theStream;
    APPListIterator<AppStream*> theIterator( *streams( ) );
    cout << endl;
    cout << "Event Output Statistics" << endl;
    cout << "=======================" << endl;
    while ( theStream = (BdbOutputStream**)theIterator( ) ) {
        cout << "Stream: " << (*theStream)->name( ) << endl
             << "     Events output: " << (*theStream)->events( ) << endl
             << "     Events owned : " << (*theStream)->ownedEvents( ) << endl
             << "     Events fully owned : " << (*theStream)->fullOwnedEvents( ) << endl;
    }

    // Dump the Objectivity statistics for EventStore, if requested
    if ( _statistics.value( ) ) {
        ooRunStatus( );
    }
    theStore->deactivate() ;

    return result;
}

void
BdbEventOutput::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 output module for the BaBar Event" << endl ;
    cout << "Store. This output module is used to store event data into an" << endl ;
    cout << "Objectivity federation whose \"Boot\" file is set using the" << endl ;
    cout << "environment variable OO_FD_BOOT." << endl ;
    cout << endl ;
    cout << "This module supports the concept of multiple output streams, each such" << endl ;
    cout << "stream corresponding to a different output collection. The 'output'" << endl ;
    cout << "command should be used to specify the streams, and the binding between" << endl ;
    cout << "the execution path(s) and output stream." << endl ;
    cout << endl;

    APPOutputModule::help( argc, argv );
}
    
//      -------------------------------------------
//      -- Protected Function Member Definitions --
//      -------------------------------------------

void
BdbEventOutput::initialize( )
{
    BdbEventStore* theStore = BdbEventStore::instance( );

    if ( ! _isPrepared ) {

        // Prepare Objectivity configuration
        prepareObjyConfig( );

        // Configure the default event store clustering
        theStore->configureClustering( );

        // Tokenize the "headers-to-be-renewed"
        delete _tokenizedHdrList ;
        _tokenizedHdrList = new vector<string>( 0 );
        char *theHdrList = new char[ strlen( _renewHdrList.value() ) + 1 ];
        strcpy( theHdrList, _renewHdrList.value() );
        if( strlen( theHdrList ) > 0 ) {
            char *theHdrName = strtok( theHdrList, " " );
            while( theHdrName != 0 ) {
                _tokenizedHdrList->push_back( theHdrName );
                theHdrName = strtok( 0, " " );
            }
            std::sort(_tokenizedHdrList->begin(),_tokenizedHdrList->end());
        }
        delete [] theHdrList;

        _isPrepared = true;

        // Loop over output streams preparing their placement strategies
        BdbOutputStream** theStream;
        APPListIterator<AppStream*> theIterator( *streams( ) );
        while ( theStream = (BdbOutputStream**)theIterator( ) ) {
            (*theStream)->setOutputModule( this );
        }       
    }
}

BdbEvsPlacementManager*
BdbEventOutput::preparePlacement( const char* const theStreamName )
{

    /* The option to write to an new set of database files is only valid
       for phase3 streams.
    */
    if(allStreamsHaveOutputTrans() == false &&
     _forceNewDatabases.value() == true){

      ErrMsg(fatal)  << "You may NOT write to a new set of" 
                     << " database files without using  <collectionName@bootfile>  (phase3)  streams" << endmsg;
      ::abort();
    }

    BdbAbstractClusteringHint* nodeHint( 0 );
    BdbAbstractClusteringHint* bridgeCollectionHint( 0 );
    BdbAbstractClusteringHint* treeCollectionHint( 0 );
    BdbAbstractClusteringHint* vectorCollectionHint( 0 );
    BdbEvsPlacementManager* thePlacement( 0 );
    BdbEventStore* theStore = BdbEventStore::instance( );

    // Determine the name to be used for this placement manager. By
    // default this is that set by defaultPlacementName( ), unless the
    // useStreamPlacement boolean flag is set true, in which case it is
    // the stream name.

    const char* nameToUse( theStreamName );
    if ( ! _useStreamPlacement.value( ) ) {
        nameToUse = BdbEvsPlacementManager::defaultPlacementName( );
    }

    char* fullName( BdbModulesUtility::placementName( nameToUse,
                                                      *theStore ) );

    // Select placement scheme based on standard or all-in-one strategy
    if ( ! _allInOne.value() ) {

        // Standard strategy
        thePlacement = createStandardPlacement( fullName );
    } else {

        // All-in-one strategy
        thePlacement = createAllInOnePlacement( fullName );

        // Set up All-in-one collections
        thePlacement->setBridgeCollectionHint( 0 );
        thePlacement->setTreeCollectionHint  ( 0 );
        thePlacement->setVectorCollectionHint( 0 );
    }

    delete [] fullName;

    return( thePlacement );
}

void
BdbEventOutput::prepareObjyConfig( )
{
    BdbTransactionClient* client( this ) ;
    BdbTransactionMgr* theMgr( transactionMgr() ) ;
    if ( 0 != theMgr ) {
       client = theMgr ;
    }

    BdbTransactionMgr::prepare( *client ) ;
}

void
BdbEventOutput::conditionalStart()
{
    // Start an update transaction for the event store (to make sure
    //   that it is active and running before the conditional Start.)
    BdbEventStore* theStore = BdbEventStore::instance( );
    BdbMode oldMode = theStore->change( BdbcUpdate );
    BdbTransactionClient* client( this ) ;
    BdbTransactionMgr* theMgr( transactionMgr() ) ;
    if ( 0 != theMgr ) {
       client = theMgr ;
    }

    BdbTransactionMgr::conditionalStart( *client ) ;

    // Restore the transaction state and force a transaction commit
    theStore->change( oldMode, d_True );
}


BdbTransactionMgr*
BdbEventOutput::transactionMgr( ) const
{
    // Locate the Transaction Manager Module if it exists and is enabled
    BdbEventOutput* self = (BdbEventOutput*) this;
    if ( 0 == _transactionMgr ) {
        self->_transactionMgr = (BdbTransactionMgr*) framework( )->fetchModule( BdbTransactionMgr::transactionMgrName( ).c_str() );
    }
    if ( 0 != _transactionMgr ) {
        if ( ! _transactionMgr->isEnabled( ) ) {
            self->_transactionMgr = 0;
        }
    }
    return _transactionMgr;
}

bool
BdbEventOutput::doEvent(AbsEvent* anEvent ) 
{
    bool result = false;
    if ( _verbose.value( ) ) {
        ErrMsg( routine ) << name( )
                        << ": doEvent( ) started at "
                        << BdbTimeStamp( )
                        << endmsg;
    }

    BdbEventStore* theStore( BdbEventStore::instance() ) ;
    theStore->activate() ;

    // Start an update transaction if not already in one
    theStore->change( BdbcUpdate );

    // Bypass all processing if the BDBDISABLEOUTPUT environment
    // variable is set.  This is a diagnostic tool for performance
    // scaling tests.

    if ( 0 == getenv( "BDBDISABLEOUTPUT" ) ) {

        // Loop over output streams performing output as appropriate
        BdbOutputStream** theStream;
        bool status;
        APPListIterator<AppStream*> theIterator( *streams( ) );
        while ( theStream = (BdbOutputStream**)theIterator( ) ) {
            status = (*theStream)->output( anEvent, this );
            if ( status ) {
                result = status;
            }
        }
    } else {
        if ( _verbose.value( ) ) {
            ErrMsg( routine ) << name( )
                            << ": Output disabled by BDBDISABLEOUTPUT environment variable"
                            << endmsg;
        }
    }
    theStore->deactivate() ;

    if ( _verbose.value( ) ) {
        if ( result ) {
            ErrMsg( routine ) << name( )
                              << ": doEvent( ) completed with output at "
                              << BdbTimeStamp( )
                              << endmsg;
        } else {
            ErrMsg( routine ) << name( )
                              << ": doEvent( ) completed without output at "
                              << BdbTimeStamp( )
                              << endmsg;
        }
    }
    return result;
}

void
BdbEventOutput::doEvent( AbsEvent* anEvent, 
                         BdbAbsCollectionT<BdbEventT>* theCollection, 
                         BdbEvsPlacementManager* thePlacement, 
                         bool buildNewEvent)
{
    BdbHandle(BdbEvent) outputEvent( 0 );
    BdbHandle(BdbEvent) oldOutputEvent( 0 ); // if one exists
    BdbHandle(BdbEvent) inputEvent( 0 );
    BdbAbstractClusteringHint* eventHint;
    bool rewritePersistent( false ) ;
    bool status;

    BdbRef(BdbEvent)* theOldPEv
      = Ifd< BdbRef(BdbEvent) >::get( anEvent, BdbEventOutput::newBdbEventKey() );
    if ( 0 != theOldPEv ) {

        oldOutputEvent = *theOldPEv;
        if(buildNewEvent == true)oldOutputEvent = 0;
    }




    if ( _verbose.value( ) ) {
        ErrMsg( routine ) << name( )
                        << ": internal doEvent( ) called from output stream at "
                        << BdbTimeStamp( )
                        << endmsg;
    }

    // Prepare the clustering hints within the placement manager
    thePlacement->prepareHints() ;

    if ( _verbose.value( ) ) {
        ErrMsg( routine ) << name( )
                        << ": prepareHints( ) finished at "
                        << BdbTimeStamp( )
                        << endmsg;
    }

    // Cache the Event clustering hint locally for quick access.
    const vector< BdbAbstractClusteringHint* >* eventPolicy( thePlacement->getPolicyHints( string( "evt" ) ) ) ;
    if ( ( 1 != (*eventPolicy).size() ) ||
       ( 0 == (*eventPolicy)[ 0 ] ) ) {
        ErrMsg( fatal ) << name() 
                        << ": No ClusteringHint for the Event"
                        << endmsg;
        ::abort() ;
    }
    eventHint = (*eventPolicy)[ 0 ] ;

    // Locate the input persistent event within the transient one
    BdbEventT* inputEventT = Ifd< BdbEventT >::get( anEvent, 
                                                    BdbEventInput::bdbEventKey() );
    if ( 0 != inputEventT ) {
        inputEvent = ((BdbRef( BdbEvent )&)(inputEventT->getPersistent())) ;
    }
    if ( BdbIsNull( inputEvent ) ) {

        // Persistent Event doesn't exist - this means that the BdbEventInput 
        // module is not in use since otherwise the persistent event would 
        // already exist.

        // Setup the event id for the persistent event, taking it from the 
        // event id in the transient event

        AbsEventID* theAbsID;
        theAbsID= Ifd<AbsEventID>::get( anEvent, BdbEventInput::absEventIDKey() );

        
        if ( _verbose.value() ) {
          ostream& err = ErrMsg(routine);
          err << name() << "::doEvent storing eventID of: " << endl;
          err << *theAbsID << endmsg;
        }

        if ( 0 ==  theAbsID ) {
          ErrMsg(fatal) << "No AbsEventID available to put in the persistent "
                        << "event.  Fatal configuration error."
                        << endmsg;
        }

        // Now build the new persistent event
        if( BdbOptNoEventStorage::isSet( *(BdbEventStore::instance()) ) ) {
           ErrMsg( fatal ) << "Not allowed to store Events in this"
                           << " federation, so new event can not be created."
                           << endmsg ;
           ::exit( -1 );
        }

        BdbRefAny theHint = eventHint->hint() ;
        outputEvent = new( theHint ) BdbEvent_001( *theAbsID );
        
        // See if there are any additional event ids to save
        HepAList< AbsEventID >* eidList( 0 ) ;
        eidList = Ifd< HepAList< AbsEventID > >::get( anEvent ,
                                                      BdbEventInput::absEventIDListKey() ) ;

        // If there is a list of secondary event ids copy those into
        // the persistent event.
        if( ( 0 != eidList ) &&
            ( 1 < eidList->length() ) ){
           const size_t finishedEids( eidList->length() ) ;
           for( size_t eid( 1 ) ;
                finishedEids != eid ;
                eid++ ) {
              outputEvent->addEventID( *( (*eidList)[ eid ] ) ) ;
           }
        }

        // Add the event to the collection. Note that we load the appropriate
        // vector collection clustering hint in case a new vector collection
        // has to be created in order to hold this event.
        BdbVectorCollectionP::clustering.setDelegate( thePlacement->vectorCollectionHint( ) );
        BdbEventT outputEventT( outputEvent );
        theCollection->add( outputEventT );


        if ( _verbose.value( ) ) {
            ErrMsg( routine ) << name( )
                            << ": persistent event created at "
                            << BdbTimeStamp( )
                            << endmsg;
        }
    } else {

        // The persistent event already exists - it must have come from 
        // BdbEventInput.  Check whether the output collection is the same as 
        // the input collection.  Add the event to the output collection
        //  if necessary.

        BdbAbsCollectionT<BdbEventT>* theInputColl = 
        Ifd<BdbAbsCollectionT<BdbEventT> >::get( anEvent, BdbEventInput::inputCollectionKey() );
        if ( ( 0 == theInputColl ) || ( *theCollection != *theInputColl ) ) {

            // Input & Output collections are different - create a new persistent event
            // based upon the original input persistent event.
            if( ! _skimEvent.value() ) {

                // Clone the event:  create a new persistent event and add
                // it to the output collection.
               if( BdbOptNoEventStorage::isSet( *(BdbEventStore::instance()) ) ) {
                  ErrMsg( fatal ) << "No longer allowed to store Events in"
                                  << " this federation, so new event can"
                                  << " not be created."
                                  << endmsg ;
                  ::exit( -1 );
               }

                BdbRefAny theHint = eventHint->hint() ;


                if(false == buildNewEvent){

                     outputEvent = new( theHint ) BdbEvent_001( inputEvent ,
                                                         *_tokenizedHdrList );


                     // See if there are any additional event ids to save
                     HepAList< AbsEventID >* eidList( 0 ) ;
                     eidList = Ifd< HepAList< AbsEventID > >::get( anEvent ,
                     BdbEventInput::absEventIDListKey() ) ;
                // If there is a list of secondary event ids beyond those in
                // the original event copy those into the new persistent event.
                
                     const size_t originalEidCount(inputEvent->eventIDCount());
                     if( ( 0 != eidList ) &&
                     ( originalEidCount < eidList->length() ) ){
                        const size_t finishedEids( eidList->length() ) ;
                        for( size_t eid( originalEidCount ) ;
                        finishedEids != eid ;
                        eid++ ) {
                           outputEvent->addEventID( *( (*eidList)[ eid ] ) ) ;
                        }
                     }
                }else{
                     AbsEventID* theAbsID;
                     theAbsID= Ifd<AbsEventID>::get( 
                     anEvent, BdbEventInput::absEventIDKey() );
                     outputEvent = new( theHint ) BdbEvent_001( *theAbsID );
                }



                

                // Add the event to the collection. Note that we load
                // the appropriate vector collection clustering hint in
                // case a new vector collection has to be created in
                // order to hold this event.
                BdbVectorCollectionP::clustering.setDelegate( thePlacement->vectorCollectionHint( ) );

                // If AllInOne option or reclusteriong option, then
                // allow persistent objects to be rewritten.
                if ( ( _allInOne.value() ) ||
                     ( _reclustering.value() ) ) {
                   rewritePersistent = true ;
                }
            } else {

                if(buildNewEvent == true){

                   ErrMsg( fatal ) << name( )
                            << ": Attempted to skim when outputting to "
                              << " stream that has separate transaction! "
                              << endmsg;
                   ::abort( );


                }
                // Skim the event: add the reference to the existing event
                // to the output collection.
                outputEvent = inputEvent ;
            }
            BdbEventT outputEventT( outputEvent );
            theCollection->add( outputEventT );

            BdbHandle( BdbEventTag ) tagH;
            BdbStatus status = outputEvent->tag( tagH ) ;
            
            // If Persistent tag exists then it is borrowed from
            // input event, and must copy descriptor
            if ( ! BdbIsNull( tagH ) ) {
               copyTagDescriptor( theCollection ,
                                  *inputEventT ) ;
            }
        }
    }

    if( inputEvent != outputEvent ) {

       // Add the persistent output event back into the transient event
       // for future use. First check to see whether one already exists.
       // If it does, it must be for a previous output stream and
       // the current one has output forced. Avoid a memory leak by
       // reusing the existing pointer.
        BdbRef(BdbEvent)* thePtr = Ifd< BdbRef(BdbEvent) >::get( anEvent,
                                   BdbEventOutput::newBdbEventKey() );
        if ( 0 == thePtr ) {
            BdbRef(BdbEvent)* thePtr = new BdbRef(BdbEvent)( outputEvent );
            status = Ifd< BdbRef(BdbEvent) >::put( anEvent,
                                              thePtr,
                                              newBdbEventKey() );
            assert( status );
        } else {
            *thePtr = outputEvent;
        }
    }


    BdbStateIdT stateId;
    if ( BdbcSuccess != outputEvent->getStateId(stateId) ) {
        // try to find it in Ifd
        vector<BdbStateIdT>* stateIDList;
        stateIDList = Ifd< vector<BdbStateIdT> >::get( anEvent, BdbEventInput::bdbStateIdListKey() );
        if ( 0 != stateIDList ) {
            if ( 1 < stateIDList->size() ) {
                 ErrMsg( fatal ) << "This version supports only one stateId. " << endmsg;
            }
            stateId = (*stateIDList)[0];
            COUT2 << "BdbEventOutput: Using stateId from Ifd: " << stateId << endl;
        } else {
            BdbTime t = outputEvent->eventID()->condKeyTriplet().key();
            stateId.setFromDefault(t);
            COUT2 << "BdbEventOutput: Using default stateId: " << stateId << endl;
        }
        outputEvent->setStateId(stateId);
    }


    // Put the output event collection into the event so that it's
    // available to the other Modules. Note that this collection 
    // is only valid for the current ouput stream, and in order to
    // avoid memory leaks, we must first check for the existence
    // of such an insertion from a prior stream. 
    // Note also that a new transient collection bound to the existing 
    // persistent collection is created since it's deleted when the 
    // transient event is deleted.
    // Locate the collection that the keys are going to be associated with
    BdbAbsCollectionT<BdbEventT>* theColl( 0 ) ;
    theColl = Ifd< BdbAbsCollectionT< BdbEventT > >::get( anEvent, 
                                                        outputCollectionKey() );
    if ( 0 != theColl ) {

        // A version of the transient collection already exists.
        // This must be from a previous output stream.
        // Replace the transient collection with that for this 
        // stream.
       BdbPtrProxy< BdbAbsCollectionT< BdbEventT > >* theCollProxy( 0 );
       theCollProxy = Ifd< BdbPtrProxy< BdbAbsCollectionT< BdbEventT > > >::get( anEvent,
                                                                               collectionProxy ) ;
       // assert if theCollProxy does not exist, but outputCollection does
       assert( 0 != theCollProxy );

       theCollProxy->setPtr( theCollection );
    } else {

        // No previous output collection exists in the event
        // Create one bound to the same persistent collection and
        // insert that into the event.
       BdbPtrProxy< BdbAbsCollectionT< BdbEventT > >* theCollProxy( new BdbPtrProxy< BdbAbsCollectionT< BdbEventT > >( theCollection ) ) ;
       status  = Ifd< BdbAbsCollectionT< BdbEventT > >::put( anEvent,
                                                           theCollProxy,
                                                           outputCollectionKey() );
       assert( status );
       BdbPtrProxy< BdbPtrProxy< BdbAbsCollectionT< BdbEventT > > >* theProxyProxy( new BdbPtrProxy< BdbPtrProxy< BdbAbsCollectionT< BdbEventT > > >( theCollProxy ) );
       status  = Ifd< BdbPtrProxy< BdbAbsCollectionT< BdbEventT > > >::put( anEvent,
                                                                          theProxyProxy,
                                                                          collectionProxy );
       assert( status );
    }

    // Get conversion manager
    BdbConversionManager* conversionManager
       = Ifd< BdbConversionManager >::get( anEvent );
    BdbHandle( BdbPersObj ) eventOwner ;
    outputEvent->owner( eventOwner ) ;
    if( theCollection->persistent() == eventOwner ) {
           
       if ( 0 == conversionManager ) {
          ErrMsg( error ) << name()
                          << ": failed to find ConversionManager in Event.\n"
                          << "No output for this event."
                          << endmsg ;
       }
       else
       {
          // Now actually tell converter and conversion manager to initiate
          // output
          if ( _verbose.value( ) ) {
             ErrMsg( routine ) << name( )
                               << ": about to call conversionManager->convertToPersistent( ) at "
                               << BdbTimeStamp( )
                               << endmsg;
          }
          if( conversionManager->outputRequired() ) {
//           BdbEventT eventT( outputEvent ) ;
             BdbEventT eventT( outputEvent ,
                               getHeaderKeys( theCollection->persistent() ) ) ;
             conversionManager->convertToPersistent( *anEvent ,
                                                     *thePlacement ,
                                                     eventT ,
                                                     rewritePersistent ) ;
             // If there was an old output event this means we may
             // not have a complete output event. Therefore borrow,
             // from the preceding output event, any data which has
             // not been written into the current output event.
             if ( !BdbIsNull( oldOutputEvent ) ) {
                ErrMsg( routine ) << "Borrowing any missing data from event"
                                  << BdbPrintOID(oldOutputEvent)
                                  << " for event "
                                  << BdbPrintOID(outputEvent)
                                  << "."
                                  << endmsg;
                eventT.borrowAsNecessary( BdbEventT( oldOutputEvent ) ) ;
             }
          }
       }
    }
    else {
       if ( ( 0 != conversionManager ) &&
              ( conversionManager->outputRequired() ) ) {
          ErrMsg( error ) << name()
                          << ": The output collection is not the owner"
                          << " of this event."
                          << " But there has been a request for output, "
                          << " which will be ignored."
                          << endmsg;
       }
    }
}

bool
BdbEventOutput::conditionalCommit( BdbCheckPointFunc& func )
{
    BdbTransactionClient* client( this ) ;
    BdbTransactionMgr* theMgr( transactionMgr() ) ;
    if ( 0 != theMgr ) {
       client = theMgr ;
    }

    bool result( BdbTransactionMgr::conditionalCheckPoint( *this ,
                                                           *client ,
                                                           func ) ) ;

    if (result ) {

        // Display the accumulated statistics if appropriate
        if ( _commitStatistics.value( ) ) {

            // Loop over output streams printing their statistics information
            BdbOutputStream** theStream;
            APPListIterator<AppStream*> theIterator( *streams( ) );
            cout << name( ) 
                 << " Transaction Committed Statistics at " 
                 << BdbTimeStamp( ) << endl;
            while ( theStream = (BdbOutputStream**)theIterator( ) ) {
                cout << "    Stream: " << (*theStream)->name( )
                     << " " << (*theStream)->events( )
                     << " Events output (" << (*theStream)->ownedEvents( ) << " owned)"
                     << endl;
            }
        }
    }
    return result;
}

void
BdbEventOutput::doCommit()
{
    BdbEventStore* theStore = BdbEventStore::instance( );

    // Make sure EventStore is active (can be called from
    //   modules not knowing anything about Objy!)
    theStore->activate() ;
    if ( theStore->transIsOpen() ) {

        // This will close any transaction started in 'doEvent'
        theStore->change( BdbcNoOpen ) ;
        if ( theStore->transIsOpen() ) {
            ErrMsg( fatal ) << name( )
                            << ": transaction commit failed at "
                              << BdbTimeStamp( )
                              << " - Event: "
                              << _total
                              << endmsg;
            ::abort( );
        }
        if ( _verbose.value( ) ) {
            ErrMsg( routine ) << name( )
                              << ": transaction committed at "
                              << BdbTimeStamp( )
                              << " - Event: "
                              << _total
                              << endmsg;
        }
    }
    theStore->deactivate() ;
}

void
BdbEventOutput::updateWhenAvailable( )
{
    BdbEventStore* theStore = BdbEventStore::instance( );
    theStore->change( BdbcUpdate ) ;
}

//              -----------------------------------------
//              -- Private Function Member Definitions --
//              -----------------------------------------

BdbEvsPlacementManager*

BdbEventOutput::createStandardPlacement( const char* theName )

{
   // Create the `Data' hints, i.e. Event, Tag and Components.
   //   NOTE: Event component must be first!
   const int kNumberComponents( 11 ) ;
   string kComponentsName[] = {
      "evt" ,
      "tag" ,
      "tru" ,
      "sim" ,
      "raw" ,
      "rec" ,
      "esd" ,
      "aod" ,
      "usr" ,
      "usrhdr" ,
      "evshdr" ,
   } ;

   vector< string* > components( kNumberComponents ) ;
   for( int i = 0 ;
        kNumberComponents != i ;
        i++ ) {
      components[ i ] = &(kComponentsName[ i ]) ;
   }

   // Build up the policies using the `Data' hints
   // gg fixed this for HP/aCC on 15/10/98
   const int kNumberDataPolicies = 88 ;
   const int kDataLabelOffset = 1 ;
   const int kMaxDataPolicyComponents = 3 ;
   string kDataPolicyDescription[ kNumberDataPolicies ][ kDataLabelOffset + kMaxDataPolicyComponents ] =
   {
      { "evt" , "evt" , "" , "" } ,
      { "tag" , "tag" , "" , "" } ,
      { "tru" , "tru" , "" , "" } ,
      { "sim" , "sim" , "" , "" } ,
      { "raw" , "raw" , "" , "" } ,
      { "rec" , "rec" , "" , "" } ,
      { "esd" , "esd" , "" , "" } ,
      { "aod" , "aod" , "" , "" } ,
      { "usr" , "usr" , "" , "" } ,
      { "astmap2pcb" ,                  "rec" , "" , "" } ,
      { "astmap2stripgtrackp" ,         "sim" , "" , "" } ,
      { "astmatchpbt" ,                 "rec" , "" , "" } ,
      { "astmatchptb" ,                 "rec" , "" , "" } ,
      { "btacandidatep" ,               "rec" , "rec" , "" } ,
      { "btacandp_001",                 "aod" , "" , "" } ,
      { "btamicrocandp" ,               "aod" , "" , "" } ,
      { "bunchlistp" ,                  "rec" , "" , "" } ,
      { "dchcostimelistp" ,             "raw" , "" , "" } ,
      { "dchdigilistp" ,                "raw" , "" , "" } ,
      { "dchdigimclistp" ,              "sim" , "" , "" } ,
      { "dchdigiwflistp" ,              "raw" , "" , "" } ,
      { "dchghitlistp" ,                "sim" , "" , "" } ,
      { "dchhitlistp" ,                 "rec" , "" , "" } ,
      { "dchpidinfop" ,                 "rec" , "" , "" } ,
      { "drcdigilistp" ,                "raw" , "" , "" } ,
      { "drcghitlistp" ,                "sim" , "" , "" } ,
      { "drcgtrkhitlistp" ,             "sim" , "" , "" } ,
      { "drcpidinfop" ,                 "rec" , "" , "" } ,
      { "drcminitrkp" ,                 "esd" , "" , "" } ,
      { "drcrecohitlistp" ,             "rec" , "" , "" } ,
      { "drcsimhitlistp" ,              "sim" , "" , "" } ,
      { "drctrackp" ,                   "rec" , "" , "" } ,
      { "eideventcontextp" ,            "raw" , "" , "" } ,
      { "emcbumpp" ,                    "rec" , "" , "" } ,
      { "emccalibdigilistp" ,           "rec" , "" , "" } ,
      { "emccandp" ,                    "rec" , "" , "" } ,
      { "emcclusterp" ,                 "rec" , "" , "" } ,
      { "emcdigilistp" ,                "raw" , "" , "" } ,
      { "emcghitlistp" ,                "sim" , "" , "" } ,
      { "emcshareddigilistp" ,          "rec" , "" , "" } ,
      { "emcwaveformlistp" ,            "raw" , "" , "" } ,
      { "geventlistp" ,                 "tru" , "" , "" } ,
      { "gtracklistp" ,                 "tru" , "" , "" } ,
      { "gvertexlistp" ,                "tru" , "" , "" } ,
      { "ifr1dclusterlistp" ,           "rec" , "" , "" } ,
      { "ifr2dclusterp" ,               "rec" , "" , "" } ,
      { "ifrabs3dp" ,                   "rec" , "" , "" } ,
      { "ifrdigistriplistp" ,           "raw" , "" , "" } ,
      { "ifrdigitimelistp" ,            "raw" , "" , "" } ,
      { "ifrfedatumlistp" ,             "raw" , "" , "" } ,
      { "ifrghitlistp" ,                "sim" , "" , "" } ,
      { "ifrinner1dclusterlistp" ,      "rec" , "" , "" } ,
      { "ifrpidinfop" ,                 "esd" , "" , "" } ,
      { "ifrtdcdatumlistp" ,            "raw" , "" , "" } ,
      { "l1dbltdigip" ,                 "raw" , "" , "" } ,
      { "l1dptddigilistp" ,             "raw" , "" , "" } ,
      { "l1dtsfdigilistp" ,             "raw" , "" , "" } ,
      { "l1emterrordigip" ,             "raw" , "" , "" } ,
      { "l1emtgltdigip" ,               "raw" , "" , "" } ,
      { "l1emtphidigilistp" ,           "raw" , "" , "" } ,
      { "l1fctdigip" ,                  "raw" , "" , "" } ,
      { "l1gltdigip" ,                  "raw" , "" , "" } ,
      { "l1gltinputspp" ,               "raw" , "" , "" } ,
      { "l3dr0p" ,                      "raw" , "" , "" } ,
      { "l3dtrklistp" ,                 "raw" , "" , "" } ,
      { "l3eclusterlistp" ,             "raw" , "" , "" } ,
      { "l3ephiclusterlistp" ,          "raw" , "" , "" } ,
      { "l3emtdecisionp" ,              "raw" , "" , "" } ,
      { "l3minputlinearrayp" ,          "raw" , "" , "" } ,
      { "l3mscriptflagarrayp" ,         "raw" , "" , "" } ,
      { "l3terrorflagp" ,               "raw" , "" , "" } ,
      { "microstdhepp" ,                "aod" , "" , "" } ,
      { "neutralhadp" ,                 "rec" , "" , "" } ,
      { "oepftransitionp" ,             "raw" , "" , "" } ,
      { "oeplbinnedprescalerrecordp" ,  "raw" , "" , "" } ,
      { "oeplinearrayinputp" ,          "raw" , "" , "" } ,
      { "oeplinearrayoutputp" ,         "raw" , "" , "" } ,
      { "pepcollisionp" ,               "tru" , "" , "" } ,
      { "rdsitemlistp" ,                "rec" , "" , "" } ,
      { "stdhepp" ,                     "tru" , "" , "" } ,
      { "svtdigilistp" ,                "raw" , "" , "" } ,
      { "svtghitlistp" ,                "sim" , "" , "" } ,
      { "svtpidinfop" ,                 "rec" , "" , "" } ,
      { "svtrootclusterlistp" ,         "rec" , "" , "" } ,
      { "trkidmanagerp" ,               "rec" , "" , "" } ,
      { "trkkalminicompositep",         "esd" , "" , "" } ,
      { "trkshellp" ,                   "rec", "rec" , "rec" },
      { "trkrecotrkproxyp" ,            "rec" }
   } ;
   
   vector< vector< string* >* > dataPolicies( kNumberDataPolicies ) ;
   for( int j = 0 ;
        kNumberDataPolicies != j ;
        j++ ) {

      // Count the number of entries in this policy.
      int numberEntries( 0 ) ;
      while ( ( numberEntries != (kDataLabelOffset + kMaxDataPolicyComponents) ) &&
              ( "" != kDataPolicyDescription[ j ][ numberEntries ] ) ) {
         ++numberEntries ;
      }

      if ( kDataLabelOffset >= numberEntries ) {
         ErrMsg( fatal ) << name() 
                         << ": Data Policy with index "
                         << j
                         << " does not have enough components (min is 1)"
                         << endmsg;
         ::abort() ;
      }

      vector< string* >* dataPolicy = new vector< string* >( numberEntries ) ;
      for ( int jj( 0 ) ;
            numberEntries != jj ;
            jj++ ) {
         (*dataPolicy)[ jj ] = &(kDataPolicyDescription[ j ][ jj ]) ;
      }
      dataPolicies[ j ] = dataPolicy ;
   }

   // Build up the policies using the `Header' hints
   const int kNumberHeaderPolicies = 16 ;
   const int kHeaderLabelOffset = 1 ;
   const int kMaxHeaderPolicyComponents = 1 ;
   string kHeaderPolicyDescription[ kNumberHeaderPolicies ][ kHeaderLabelOffset + kMaxHeaderPolicyComponents ] =
   {
      { "tru" , "tru" } ,
      { "sim" , "sim" } ,
      { "raw" , "raw" } ,
      { "usr" , "usrhdr" } ,
      { "cmp" , "evshdr" } , // composite BtaCandidates
      { "bta" , "evshdr" } ,
      { "dch" , "evshdr" } ,
      { "drc" , "evshdr" } ,
      { "emc" , "evshdr" } ,
      { "ifr" , "evshdr" } ,
      { "svt" , "evshdr" } ,
      { "trk" , "evshdr" } ,
      { "umc" , "evshdr" } ,
      { "rec" , "evshdr" } ,
      { "pid" , "evshdr" } ,
      { "trg" , "evshdr" }
   } ;
   
   vector< vector< string* >* > headerPolicies( kNumberHeaderPolicies ) ;
   for( int k = 0 ;
        kNumberHeaderPolicies != k ;
        k++ ) {

      // Count the number of entries in this policy.
      int numberEntries( 0 ) ;
      while ( ( numberEntries != (kHeaderLabelOffset + kMaxHeaderPolicyComponents) ) &&
              ( "" != kHeaderPolicyDescription[ k ][ numberEntries ] ) ) {
         ++numberEntries ;
      }

      if ( kHeaderLabelOffset >= numberEntries ) {
         ErrMsg( fatal ) << name() 
                         << ": Header Policy with index "
                         << k
                         << " does not have enough components (min is 1)"
                         << endmsg;
         ::abort() ;
      }

      vector< string* >* headerPolicy = new vector< string* >( numberEntries ) ;
      for ( int kk( 0 ) ;
            numberEntries != kk ;
            kk++ ) {
         (*headerPolicy)[ kk ] = &(kHeaderPolicyDescription[ k ][ kk ]) ;
      }
      headerPolicies[ k ] = headerPolicy ;
   }

   BdbEventStore* theStore = BdbEventStore::instance( );
   BdbEvsPlacementManager* placement = new BdbEvsPlacementManager( *theStore ,
                                                                   theName , 
                                                                   components ,
                                                                   dataPolicies ,
                                                                   headerPolicies, _forceNewDatabases.value() ) ;
   for( int m = 0 ;
        kNumberDataPolicies != m ;
        m++ ) {

      // cast away const to allow deletion.
      delete dataPolicies[ m ] ;
   }

   for( int n = 0 ;
        kNumberHeaderPolicies != n ;
        n++ ) {

      // cast away const to allow deletion.
      delete headerPolicies[ n ] ;
   }

   return ( placement ) ;
}

BdbEvsPlacementManager*

BdbEventOutput::createAllInOnePlacement( const char* theName )
{
   // Create the `Data' hints, i.e. Event, Tag and Components.
   //   NOTE: Event component must be first!
   const int kNumberComponents( 1 ) ;
   string kComponentsName[] = {
      "aio"
   } ;

   vector< string* > components( kNumberComponents ) ;
   for( int i = 0 ;
        kNumberComponents != i ;
        i++ ) {
      components[ i ] = &(kComponentsName[ i ]) ;
   }

   // Build up the policies using the `Data' hints
   const int kNumberDataPolicies = 85 ;
   const int kDataLabelOffset = 1 ;
   const int kMaxDataPolicyComponents = 3 ;
   string kDataPolicyDescription[ kNumberDataPolicies ][ kDataLabelOffset + kMaxDataPolicyComponents ] =
   {
      { "evt" , "aio" , "" , "" } ,
      { "tag" , "aio" , "" , "" } ,
      { "tru" , "aio" , "" , "" } ,
      { "sim" , "aio" , "" , "" } ,
      { "raw" , "aio" , "" , "" } ,
      { "rec" , "aio" , "" , "" } ,
      { "esd" , "aio" , "" , "" } ,
      { "aod" , "aio" , "" , "" } ,
      { "usr" , "aio" , "" , "" } ,
      { "astmap2pcb" ,                  "aio" , "" , "" } ,
      { "astmap2stripgtrackp" ,         "aio" , "" , "" } ,
      { "astmatchpbt" ,                 "aio" , "" , "" } ,
      { "astmatchptb" ,                 "aio" , "" , "" } ,
      { "btacandidatep" ,               "aio" , "aio" , "" } ,
      { "btacandp_001" ,                "aio" , "" , "" } ,
      { "btamicrocandp" ,               "aio" , "" , "" } ,
      { "bunchlistp" ,                  "aio" , "" , "" } ,
      { "dchcostimelistp" ,             "aio" , "" , "" } ,
      { "dchdigilistp" ,                "aio" , "" , "" } ,
      { "dchdigimclistp" ,              "aio" , "" , "" } ,
      { "dchdigiwflistp" ,              "aio" , "" , "" } ,
      { "dchghitlistp" ,                "aio" , "" , "" } ,
      { "dchhitlistp" ,                 "aio" , "" , "" } ,
      { "dchpidinfop" ,                 "aio" , "" , "" } ,
      { "drcdigilistp" ,                "aio" , "" , "" } ,
      { "drcghitlistp" ,                "aio" , "" , "" } ,
      { "drcgtrkhitlistp" ,             "aio" , "" , "" } ,
      { "drcpidinfop" ,                 "aio" , "" , "" } ,
      { "drcrecohitlistp" ,             "aio" , "" , "" } ,
      { "drcsimhitlistp" ,              "aio" , "" , "" } ,
      { "drctrackp" ,                   "aio" , "" , "" } ,
      { "eideventcontextp" ,            "aio" , "" , "" } ,
      { "emcbumpp" ,                    "aio" , "" , "" } ,
      { "emccalibdigilistp" ,           "aio" , "" , "" } ,
      { "emccandp" ,                    "aio" , "" , "" } ,
      { "emcclusterp" ,                 "aio" , "" , "" } ,
      { "emcdigilistp" ,                "aio" , "" , "" } ,
      { "emcghitlistp" ,                "aio" , "" , "" } ,
      { "emcshareddigilistp" ,          "aio" , "" , "" } ,
      { "emcwaveformlistp" ,            "aio" , "" , "" } ,
      { "geventlistp" ,                 "aio" , "" , "" } ,
      { "gtracklistp" ,                 "aio" , "" , "" } ,
      { "gvertexlistp" ,                "aio" , "" , "" } ,
      { "ifr1dclusterlistp" ,           "aio" , "" , "" } ,
      { "ifr2dclusterp" ,               "aio" , "" , "" } ,
      { "ifrabs3dp" ,                   "aio" , "" , "" } ,
      { "ifrdigistriplistp" ,           "aio" , "" , "" } ,
      { "ifrdigitimelistp" ,            "aio" , "" , "" } ,
      { "ifrfedatumlistp" ,             "aio" , "" , "" } ,
      { "ifrghitlistp" ,                "aio" , "" , "" } ,
      { "ifrinner1dclusterlistp" ,      "aio" , "" , "" } ,
      { "ifrpidinfop" ,                 "aio" , "" , "" } ,
      { "ifrtdcdatumlistp" ,            "aio" , "" , "" } ,
      { "l1dbltdigip" ,                 "aio" , "" , "" } ,
      { "l1dptddigilistp" ,             "aio" , "" , "" } ,
      { "l1dtsfdigilistp" ,             "aio" , "" , "" } ,
      { "l1emterrordigip" ,             "aio" , "" , "" } ,
      { "l1emtgltdigip" ,               "aio" , "" , "" } ,
      { "l1emtphidigilistp" ,           "aio" , "" , "" } ,
      { "l1fctdigip" ,                  "aio" , "" , "" } ,
      { "l1gltdigip" ,                  "aio" , "" , "" } ,
      { "l1gltinputspp" ,               "aio" , "" , "" } ,
      { "l3dr0p" ,                      "aio" , "" , "" } ,
      { "l3dtrklistp" ,                 "aio" , "" , "" } ,
      { "l3eclusterlistp" ,             "aio" , "" , "" } ,
      { "l3ephiclusterlistp" ,          "aio" , "" , "" } ,
      { "l3emtdecisionp" ,              "aio" , "" , "" } ,
      { "l3minputlinearrayp" ,          "aio" , "" , "" } ,
      { "l3mscriptflagarrayp" ,         "aio" , "" , "" } ,
      { "l3terrorflagp" ,               "aio" , "" , "" } ,
      { "microstdhepp" ,                "aio" , "" , "" } ,
      { "neutralhadp" ,                 "aio" , "" , "" } ,
      { "oepftransitionp" ,             "aio" , "" , "" } ,
      { "oeplbinnedprescalerrecordp" ,  "aio" , "" , "" } ,
      { "oeplinearrayinputp" ,          "aio" , "" , "" } ,
      { "oeplinearrayoutputp" ,         "aio" , "" , "" } ,
      { "pepcollisionp" ,               "aio" , "" , "" } ,
      { "rdsitemlistp" ,                "aio" , "" , "" } ,
      { "stdhepp" ,                     "aio" , "" , "" } ,
      { "svtdigilistp" ,                "aio" , "" , "" } ,
      { "svtghitlistp" ,                "aio" , "" , "" } ,
      { "svtpidinfop" ,                 "aio" , "" , "" } ,
      { "svtrootclusterlistp" ,         "aio" , "" , "" } ,
      { "trkidmanagerp" ,               "aio" , "" , "" } ,
      { "trkshellp" ,                   "aio", "aio" , "aio" }
   } ;
   
   vector< vector< string* >* > dataPolicies( kNumberDataPolicies ) ;
   for( int j = 0 ;
        kNumberDataPolicies != j ;
        j++ ) {

      // Count the number of entries in this Policy.
      int numberEntries( 0 ) ;
      while ( ( numberEntries != (kDataLabelOffset + kMaxDataPolicyComponents) ) &&
              ( "" != kDataPolicyDescription[ j ][ numberEntries ] ) ) {
         ++numberEntries ;
      }

      if ( kDataLabelOffset >= numberEntries ) {
         ErrMsg( fatal ) << name() 
                         << ": Data Policy with index "
                         << j
                         << " does not have enough components (min is 1)"
                         << endmsg;
         ::abort() ;
      }

      vector< string* >* dataPolicy = new vector< string* >( numberEntries ) ;
      for ( int jj( 0 ) ;
            numberEntries != jj ;
            jj++ ) {
         (*dataPolicy)[ jj ] = &(kDataPolicyDescription[ j ][ jj ]) ;
      }
      dataPolicies[ j ] = dataPolicy ;
   }

   // Build up the policies using the `Header' hints
   const int kNumberHeaderPolicies = 13 ;
   const int kHeaderLabelOffset = 1 ;
   const int kMaxHeaderPolicyComponents = 1 ;
   string kHeaderPolicyDescription[ kNumberHeaderPolicies ][ kHeaderLabelOffset + kMaxHeaderPolicyComponents ] =
   {
      { "tru" , "aio" } ,
      { "sim" , "aio" } ,
      { "raw" , "aio" } ,
      { "usr" , "aio" } ,
      { "bta" , "aio" } ,
      { "dch" , "aio" } ,
      { "drc" , "aio" } ,
      { "emc" , "aio" } ,
      { "ifr" , "aio" } ,
      { "svt" , "aio" } ,
      { "trk" , "aio" } ,
      { "umc" , "aio" } ,
      { "rec" , "aio" }
   } ;
   
   vector< vector< string* >* > headerPolicies( kNumberHeaderPolicies ) ;
   for( int k = 0 ;
        kNumberHeaderPolicies != k ;
        k++ ) {

      // Count the number of entries in this policy.
      int numberEntries( 0 ) ;
      while ( ( numberEntries != (kHeaderLabelOffset + kMaxHeaderPolicyComponents) ) &&
              ( "" != kHeaderPolicyDescription[ k ][ numberEntries ] ) ) {
         ++numberEntries ;
      }

      if ( kHeaderLabelOffset >= numberEntries ) {
         ErrMsg( fatal ) << name() 
                         << ": Header Policy with index "
                         << k
                         << " does not have enough components (min is 1)"
                         << endmsg;
         ::abort() ;
      }

      vector< string* >* headerPolicy = new vector< string* >( numberEntries ) ;
      for ( int kk( 0 ) ;
            numberEntries != kk ;
            kk++ ) {
         (*headerPolicy)[ kk ] = &(kHeaderPolicyDescription[ k ][ kk ]) ;
      }
      headerPolicies[ k ] = headerPolicy ;
   }

   BdbEventStore* theStore = BdbEventStore::instance( );
   BdbEvsPlacementManager* placement = new BdbEvsPlacementManager( *theStore ,
                                                                   theName , 
                                                                   components ,
                                                                   dataPolicies ,
                                                                   headerPolicies , _forceNewDatabases.value()) ;
   for( int m = 0 ;
        kNumberDataPolicies != m ;
        m++ ) {

      // cast away const to allow deletion.
      delete dataPolicies[ m ] ;
   }

   for( int n = 0 ;
        kNumberHeaderPolicies != n ;
        n++ ) {

      // cast away const to allow deletion.
      delete headerPolicies[ n ] ;
   }

   return( placement ) ;
}

//
// BdbTransactionClient functions
//

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

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

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

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

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

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


bool
BdbEventOutput::loggingInput() const
{
   return( false ) ;
}

bool
BdbEventOutput::loggingOutput() const
{
   return( true ) ;
}

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

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

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

size_t
BdbEventOutput::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< BdbEventOutput* >( this )->_useGranularity = valueToUse ;

   return( _useGranularity ) ;
}

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

size_t
BdbEventOutput::totalEventCount() const
{
   return( _total ) ;
}

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

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

size_t
BdbEventOutput::commitAndHoldLimit() const
{
   // Can not be managed by this client to always return zero
   return( 0 ) ;
}

size_t
BdbEventOutput::commitAndHoldCount() const
{
   return( 0 ) ;
}

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

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

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

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

size_t
BdbEventOutput::handlePrintFrequency() const 
{
   // Can not be managed by this client to always return zero
   return( 0 ) ;
}

size_t
BdbEventOutput::handlePrintCount() const 
{
   return( 0 ) ;
}

size_t
BdbEventOutput::handlePrintLevel() const 
{
   return( 0 ) ;
}

size_t
BdbEventOutput::handlePrintLimit() const 
{
   return( 0 ) ;
}

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

void
BdbEventOutput::inputLogged()
{
   ErrMsg( fatal ) << name()
                   << " can not be used as an input BdbTransactionClient"
                   << endmsg ;
   ::abort() ;
}

void
BdbEventOutput::outputLogged()
{
}

void
BdbEventOutput::incrementEventCount()
{
   _total++ ;
   _granularityCount++ ;
}

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

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

void
BdbEventOutput::incrementCommitAndHoldCount()
{
}

void
BdbEventOutput::resetCommitAndHoldCount()
{
}

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

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

void
BdbEventOutput::incrementHandlePrintCount()
{
}

void
BdbEventOutput::resetHandlePrintCount()
{
}

void
BdbEventOutput::setAllStreamsHaveOutputTrans(bool allStreamsHaveOutputTrans)
{

   _allStreamsHaveOutputTrans = allStreamsHaveOutputTrans;  
}

bool
BdbEventOutput::allStreamsHaveOutputTrans() const
{

   return _allStreamsHaveOutputTrans;
}

 

---

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

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