EventCollectionSelector.cpp

Go to the documentation of this file.

/***********************************************************************************\
* (c) Copyright 1998-2019 CERN for the benefit of the LHCb and ATLAS collaborations *
*                                                                                   *
* This software is distributed under the terms of the Apache version 2 licence,     *
* copied verbatim in the file "LICENSE".                                            *
*                                                                                   *
* In applying this licence, CERN does not waive the privileges and immunities       *
* granted to it by virtue of its status as an Intergovernmental Organization        *
* or submit itself to any jurisdiction.                                             *
\***********************************************************************************/
// Include files
#include "GaudiKernel/EventSelectorDataStream.h"
#include "GaudiKernel/GenericAddress.h"
#include "GaudiKernel/IAddressCreator.h"
#include "GaudiKernel/IDataSourceMgr.h"
#include "GaudiKernel/INTupleSvc.h"
#include "GaudiKernel/IRegistry.h"
#include "GaudiKernel/ISelectStatement.h"
#include "GaudiKernel/ISvcLocator.h"
#include "GaudiKernel/NTuple.h"
#include "GaudiKernel/ObjectFactory.h"
#include "GaudiKernel/SmartIF.h"
 
#include "EventCollectionSelector.h"
 
#include <memory>
 
DECLARE_COMPONENT( EventCollectionSelector )
 
 
class EventCollectionContext : public IEvtSelector::Context {
public:
  typedef std::list<std::string> ListName;
 
private:
  GenericAddress*                m_pAddressBuffer = nullptr;
  const EventCollectionSelector* m_pSelector      = nullptr;
  ListName                       m_files;
  std::string                    m_criteria;
  ListName::const_iterator       m_fileIterator;
  std::string                    m_currentInput;
 
public:
  EventCollectionContext( const EventCollectionSelector* pSelector );
  ~EventCollectionContext() override;
  const std::string&       currentInput() const { return m_currentInput; }
  void                     setCurrentInput( const std::string& v ) { m_currentInput = v; }
  ListName&                files() { return m_files; }
  void*                    identifier() const override { return (void*)m_pSelector; }
  void                     setCriteria( const std::string& crit ) { m_criteria = crit; }
  ListName::const_iterator fileIterator() { return m_fileIterator; }
  void                     setFileIterator( ListName::const_iterator new_iter ) { m_fileIterator = new_iter; }
};
 
EventCollectionContext::~EventCollectionContext() { m_pAddressBuffer->release(); }
 
EventCollectionContext::EventCollectionContext( const EventCollectionSelector* pSelector ) : m_pSelector( pSelector ) {
  m_pAddressBuffer = new GenericAddress();
  m_pAddressBuffer->addRef();
}
 
// IService implementation: Db event selector override
StatusCode EventCollectionSelector::initialize() {
  // Initialize base class
  StatusCode status = Service::initialize();
  if ( !status.isSuccess() ) {
    error() << "Error initializing base class Service!" << endmsg;
    return status;
  }
  m_pAddrCreator = serviceLocator()->service( "EventPersistencySvc" );
  if ( !m_pAddrCreator ) {
    error() << "Unable to locate IAddressCreator interface of "
            << "EventPersistencySvc" << endmsg;
    return status;
  }
  m_tupleSvc = serviceLocator()->service( m_tupleSvcName );
  if ( !m_tupleSvc ) {
    error() << "Unable to locate INTupleSvc interface of " << m_tupleSvcName << endmsg;
    return status;
  }
  return status;
}
 
// Connect collection to selector
StatusCode EventCollectionSelector::connectDataSource( const std::string& db, const std::string& /* typ */ ) const {
  StatusCode              status = StatusCode::FAILURE;
  SmartIF<IDataSourceMgr> svc( m_tupleSvc );
  if ( svc && !db.empty() ) {
    std::string ident = name() + ' ';
    ident += "DATAFILE='" + m_database.value().substr( 5 ) + "' ";
    if ( !m_dbSvc.empty() )
      ident += "SVC='" + m_dbSvc + "' ";
    else
      ident += "TYP='" + m_dbType + "' ";
    ident += "OPT='READ' ";
    if ( m_authentication.length() > 0 ) { ident += "AUTH='" + m_authentication + "' "; }
    status = svc->connect( ident );
  }
  return status;
}
 
StatusCode EventCollectionSelector::connectTuple( const std::string& nam, const std::string& itName,
                                                  NTuple::Tuple*& tup, NTuple::Item<IOpaqueAddress*>*& item ) const {
  std::string top    = "/NTUPLES/" + name() + '/' + nam;
  StatusCode  status = m_tupleSvc->retrieveObject( top, (DataObject*&)tup );
  if ( status.isSuccess() ) {
    item   = new NTuple::Item<IOpaqueAddress*>();
    status = tup->item( itName, *item );
    if ( status.isSuccess() ) return status;
    error() << "Item " << itName << " is not part of the collection:" << top << endmsg;
    delete item;
    item = nullptr;
  } else {
    error() << "Cannot connect to collection:" << top << endmsg;
  }
  tup = nullptr;
  return status;
}
 
StatusCode EventCollectionSelector::connectStatement( const std::string& typ, const std::string& crit,
                                                      INTuple* tuple ) const {
  std::string seltyp = typ;
  if ( !seltyp.empty() || !crit.empty() ) {
    if ( !crit.empty() && seltyp.length() == 0 ) seltyp = "NTuple::Selector";
    SmartIF<ISelectStatement> stmt( ObjFactory::create( seltyp, serviceLocator() ).release() );
    if ( stmt ) {
      if ( !crit.empty() ) stmt->setCriteria( crit );
      tuple->attachSelector( stmt ).ignore();
      return StatusCode::SUCCESS;
    }
    return StatusCode::FAILURE;
  }
  return StatusCode::SUCCESS;
}
 
StatusCode EventCollectionSelector::getNextRecord( NTuple::Tuple* tuple ) const {
  StatusCode status = StatusCode::FAILURE;
  if ( tuple ) {
    do {
      status = m_tupleSvc->readRecord( tuple );
      if ( status.isSuccess() ) {
        ISelectStatement* statement = tuple->selector();
        bool              use_it    = ( statement ) ? ( *statement )( tuple ) : true;
        if ( use_it ) { return status; }
      }
    } while ( status.isSuccess() );
  }
  return status;
}
 
StatusCode EventCollectionSelector::getPreviousRecord( NTuple::Tuple* tuple ) const {
  StatusCode status = StatusCode::FAILURE;
  if ( tuple ) {
    IRegistry* pReg = tuple->registry();
    if ( pReg ) {
      IOpaqueAddress* pAddr = pReg->address();
      if ( pAddr ) {
        long* ip = (long*)pAddr->ipar();
        do {
          if ( ip[1] > 1 ) {
            ip[1] -= 2;
            status = m_tupleSvc->readRecord( tuple );
            if ( status.isSuccess() ) {
              ISelectStatement* statement = tuple->selector();
              bool              use_it    = ( statement ) ? ( *statement )( tuple ) : true;
              if ( use_it ) { return status; }
            }
          } else {
            return StatusCode::FAILURE;
          }
        } while ( status.isSuccess() );
      }
    }
  }
  return StatusCode::FAILURE;
}
 
// Connect collection to selector
StatusCode EventCollectionSelector::connectCollection( MyContextType* ctxt ) const {
  if ( ctxt ) {
    StatusCode status = connectDataSource( m_database, m_dbType );
    if ( status.isSuccess() ) {
      status = connectTuple( m_cntName, m_itemName, ctxt->tuple, ctxt->item );
      if ( status.isSuccess() ) {
        status = connectStatement( m_statement, m_criteria, ctxt->tuple );
        if ( status.isSuccess() ) {
          *( ctxt->item ) = 0;
          return status;
        }
      }
    }
    return status;
  }
  return StatusCode::FAILURE;
}
 
// Finalize service
StatusCode EventCollectionSelector::finalize() {
  // release services
  m_pAddrCreator = nullptr;
  m_tupleSvc     = nullptr;
  return Service::finalize();
}
 
// FIXME - method below generates leak errors with sanitizer
//
// clang-format off
// Direct leak of 48 byte(s) in 2 object(s) allocated from:
//    #0 0x7f36dab12da8 in operator new(unsigned long) /afs/cern.ch/cms/CAF/CMSCOMM/COMM_ECAL/dkonst/GCC/build/contrib/gcc-8.2.0/src/gcc/8.2.0/libsanitizer/lsan/lsan_interceptors.cc:229
//    #1 0x7f36cca8cf7e in EventCollectionSelector::connectTuple(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, NTuple::Tuple*&, NTuple::Item<IOpaqueAddress*>*&) const ../GaudiCoreSvc/src/EventSelector/EventCollectionSelector.cpp:104
//    #2 0x7f36cca8c27f in EventCollectionSelector::connectCollection(EventCollectionSelector::MyContextType*) const ../GaudiCoreSvc/src/EventSelector/EventCollectionSelector.cpp:183
//    #3 0x7f36cca8c808 in EventCollectionSelector::createContext(IEvtSelector::Context*&) const ../GaudiCoreSvc/src/EventSelector/EventCollectionSelector.cpp:209
// clang-format on
//
// These leaks are currently suppressed in Gaudi/job/Gaudi-LSan.supp - remove entry there to reactivate
 
StatusCode EventCollectionSelector::createContext( Context*& refpCtxt ) const {
  refpCtxt          = nullptr;
  auto       ctxt   = std::make_unique<MyContextType>();
  StatusCode status = connectCollection( ctxt.get() );
  if ( !status.isSuccess() ) {
    error() << "Unable to connect Collection file \"" << m_database << "\"" << endmsg;
    return status;
  }
  refpCtxt = ctxt.release();
  return StatusCode::SUCCESS;
}
 
StatusCode EventCollectionSelector::next( Context& refCtxt ) const { return next( refCtxt, 1 ); }
 
StatusCode EventCollectionSelector::next( Context& refCtxt, int jump ) const {
  MyContextType* ctxt = dynamic_cast<MyContextType*>( &refCtxt );
  if ( ctxt ) {
    *( ctxt->item ) = ctxt->addressBuffer;
    StatusCode sc   = StatusCode::SUCCESS;
    for ( int i = 0; i < jump && sc.isSuccess(); ++i ) { sc = getNextRecord( ctxt->tuple ); }
    return sc;
  }
  return StatusCode::FAILURE;
}
 
StatusCode EventCollectionSelector::previous( Context& refCtxt ) const { return previous( refCtxt, 1 ); }
 
StatusCode EventCollectionSelector::previous( Context& refCtxt, int jump ) const {
  MyContextType* ctxt = dynamic_cast<MyContextType*>( &refCtxt );
  if ( ctxt ) {
    *( ctxt->item ) = ctxt->addressBuffer;
    StatusCode sc   = StatusCode::SUCCESS;
    for ( int i = 0; i < jump && sc.isSuccess(); ++i ) { sc = getPreviousRecord( ctxt->tuple ); }
    return sc;
  }
  return StatusCode::FAILURE;
}
 
StatusCode EventCollectionSelector::rewind( Context& /* refCtxt */ ) const { return StatusCode::FAILURE; }
 
StatusCode EventCollectionSelector::createAddress( const Context& refCtxt, IOpaqueAddress*& refpAddr ) const {
  const MyContextType* ctxt = dynamic_cast<const MyContextType*>( &refCtxt );
  if ( ctxt ) {
    IOpaqueAddress* pA = *( ctxt->item );
    if ( pA ) {
      IOpaqueAddress* pAddress = nullptr;
      StatusCode status = m_pAddrCreator->createAddress( pA->svcType(), pA->clID(), pA->par(), pA->ipar(), pAddress );
      if ( status.isSuccess() ) {
        refpAddr = pAddress;
        return StatusCode::SUCCESS;
      } else {
        error() << "Failed to access " << pA->par()[0] << ":" << pA->par()[1] << " SvcTyp:" << long( pA->svcType() )
                << " CLID:" << pA->clID() << endmsg;
      }
    }
  }
  return StatusCode::FAILURE;
}
 
StatusCode EventCollectionSelector::releaseContext( Context*& refCtxt ) const {
  MyContextType* ctxt = dynamic_cast<MyContextType*>( refCtxt );
  if ( ctxt ) {
    delete ctxt;
    ctxt = nullptr;
    return StatusCode::SUCCESS;
  }
  return StatusCode::FAILURE;
}
 
StatusCode EventCollectionSelector::resetCriteria( const std::string& cr, Context& refCtxt ) const {
  MyContextType* ctxt = dynamic_cast<MyContextType*>( &refCtxt );
  if ( ctxt ) {
    ctxt->criteria = cr;
    return StatusCode::SUCCESS;
  }
  return StatusCode::FAILURE;
}
 
StatusCode EventCollectionSelector::last( Context& /* refCtxt */ ) const { return StatusCode::FAILURE; }