DybTrimIO.cc

Go to the documentation of this file.

//
#include "DybTrimIO.h"
//
// now time
#include "Event/RegistrationSequence.h"
#include "Event/TemporalDataObject.h"
//
// IO
#include "DybKernel/ISaveB4TrimAesSvc.h"
#include "DybKernel/IDybStorageSvc.h"
//
// trimming
#include "GaudiKernel/IDataManagerSvc.h"
#include "DybKernel/IArchiveTrimSvc.h"
//
#include "GaudiKernel/SmartDataPtr.h"

#include "DataUtilities/DybArchiveList.h"
#include "GaudiKernel/DataSvc.h"
#include "GaudiKernel/IIncidentSvc.h"
#include "DybKernel/EventStoreIncident.h"

#include <map>


DybTrimIO::DybTrimIO(const std::string& name, ISvcLocator* pSvcLocator):
  GaudiAlgorithm(name, pSvcLocator),
  m_pSaveB4TrimAesSvc(0),
  m_pDybStorageSvc(0),
  m_pArchiveSvc(0),
  m_pTrimSvc(0)
{
  declareProperty("SaveFlag",m_saveFlag=false,"Save or not");

  declareProperty("NowLocation",
                  m_nowLocation =  DayaBay::RegistrationSequenceLocation::Default, 
                  "Location to determine current time, default: RegistrationSequence");

  declareProperty("RegSeqLocation",
                  m_regSeqLocation=DayaBay::RegistrationSequenceLocation::Default,
                  "Location to RegistrationSequence");

  declareProperty("SaveB4TrimAesSvc",
                  m_pSaveB4TrimAesSvcName="SaveB4TrimAesSvc",
                  "Name of SaveB4TrimAesSvc");

  declareProperty("DybStorageSvc",
                  m_pDybStorageSvcName="DybStorageSvc",
                  "Name of an IDybStorageSvc");

  declareProperty("ArchiveSvc",
                  m_pArchiveSvcName="EventDataArchiveSvc",
                  "Name of AES");

  declareProperty("TrimSvc",
                  m_pTrimSvcName="ArchiveTrimSvc",
                  "Name of trimming service");

}

// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * 

StatusCode DybTrimIO::initialize()
{
  //
  StatusCode status;
  debug() << " DybTrimIO initialize()" << endreq;

  if ( m_saveFlag ) {
    // SaveB4TrimAesSvc
    status = service(m_pSaveB4TrimAesSvcName,
                     m_pSaveB4TrimAesSvc);

    debug()<<"m_pSaveB4TrimAesSvc "<<m_pSaveB4TrimAesSvc<<endreq;
    if (status.isFailure()) {
      error()<<"Service [SaveB4TrimAesSvc] not found"<<endreq;
      return StatusCode::FAILURE;
    }

    // DybStorageSvc 
    status = service(m_pDybStorageSvcName,
                     m_pDybStorageSvc);
    
    debug()<<"m_pDybStorageSvc "<<m_pDybStorageSvc<<endreq;
    if (status.isFailure()) {
      error()<<"Service [DybStorageSvc] not found"<<endreq;
      return StatusCode::FAILURE;
    }
  }

  // Get archive store
  status = service(m_pArchiveSvcName,
                   m_pArchiveSvc);

  debug()<<"m_pArchiveSvc "<<m_pArchiveSvc<<endreq;
  if (status.isFailure()) {
    error()<<"Service [EventDataArchiveSvc] not found"<<endreq;
    return StatusCode::FAILURE;
  }

  // Get trimming Svc
  status = service(m_pTrimSvcName,
                   m_pTrimSvc);

  debug()<<"m_pTrimSvc "<<m_pTrimSvc<<endreq;
  if (status.isFailure()) {
    error()<<"Service [ArchiveTrimSvc] not found"<<endreq;
    return StatusCode::FAILURE;
  }

  return StatusCode::SUCCESS;
}

// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * 

/*FIXME: The approach of completely walking the data store for
  trimming every object is too slow.

StatusCode DybTrimIO::execute() 
{
  StatusCode sc;
  
  // get the data object at the "now" location
  SmartDataPtr<const DayaBay::TemporalDataObject> 
              pTemoral(eventSvc(),m_nowLocation);
  // check the pointer
  if (0 == pTemoral) {
    error()<<"Failed to retrieve TemporalDataObject from Event Store from "
           << m_nowLocation
           <<endreq;
    return StatusCode::FAILURE;
  }  

  // save before trim
  if(m_saveFlag) {
    sc = m_pSaveB4TrimAesSvc->store(m_regSeqLocation,
                                    m_pDybStorageSvc,
                                    m_pArchiveSvc,
                                    m_pTrimSvc,
                                    pTemoral->earliest());
    if(sc.isFailure()) {
      debug()<<"Failed to write"<<endreq;
      return StatusCode::FAILURE;
    }
  }

  // trim base on it
  sc = m_pTrimSvc->trim(m_pArchiveSvc,pTemoral->earliest());
  if(sc.isFailure()) {
    debug()<<"Failed to trim"<<endreq;
    return StatusCode::FAILURE;
  }

  return StatusCode::SUCCESS;
}
*/

// Alternative time-optimized implementation
StatusCode DybTrimIO::execute() 
{
  StatusCode sc;
  
  // get the data object at the "now" location
  SmartDataPtr<const DayaBay::RegistrationSequence> 
              pTemporal(eventSvc(),m_regSeqLocation);
  // check the pointer
  if (0 == pTemporal) {
    error()<<"Failed to retrieve RegistrationSequence from Event Store from "
           << m_regSeqLocation
           <<endreq;
    return StatusCode::FAILURE;
  }
  
  // save before trim
  if(m_saveFlag) {
    sc = m_pSaveB4TrimAesSvc->store(m_regSeqLocation,
                                    m_pDybStorageSvc,
                                    m_pArchiveSvc,
                                    m_pTrimSvc,
                                    pTemporal->earliest());
    if(sc.isFailure()) {
      debug()<<"Failed to write"<<endreq;
      return StatusCode::FAILURE;
    }
  }

  // Create handle to AES
  static DataSvc* pAes=0;
  if(!pAes){
    try{
      pAes=dynamic_cast<DataSvc*>(m_pArchiveSvc);
    }
    catch(...) {
      error() <<"Failed to cast AES" << endreq;
      return StatusCode::FAILURE;
    }
  }

  // Catch new data paths, index
  static std::map<std::string,DybArchiveList*> archiveMap;
  // First, catch the registration sequence archive
  if(archiveMap.empty()){
    DataObject* pObj=0;      
    StatusCode sc = pAes->retrieveObject(m_regSeqLocation,pObj);
    if(sc.isFailure()) {
      error() << "Failed to get archive for " << m_regSeqLocation << endreq;
      return StatusCode::FAILURE;
    }
    // Get Archive List
    DybArchiveList* pArchiveList=0;
    try{
      pArchiveList=dynamic_cast<DybArchiveList*>(pObj);
    }
    catch (...) {
      error() << "Failed to get DybArchiveList " << m_regSeqLocation << endreq;
      return StatusCode::FAILURE;
    }
    archiveMap[m_regSeqLocation] = pArchiveList;
    //info() << "Indexing archive list at " << m_regSeqLocation << endreq;
  }
  // Next, loop over paths in reg seq, and add to index if they are new
  IRegistrationSequence::Registrations::const_iterator regIter, 
    regEnd=pTemporal->registrations().end();
  for(regIter=pTemporal->registrations().begin(); 
      regIter != regEnd; 
      regIter++){
    const ObjectReg& objReg = (*regIter);
    if( archiveMap.find(objReg.path()) == archiveMap.end() ){
      // Found a new data path, add archive list to index
      DataObject* pObj=0;      
      StatusCode sc = pAes->retrieveObject(objReg.path(),pObj);
      if(sc.isFailure()) {
        error() << "Failed to get archive for " << objReg.path() << endreq;
        return StatusCode::FAILURE;
      }
      // Get Archive List
      DybArchiveList* pArchiveList=0;
      try{
        pArchiveList=dynamic_cast<DybArchiveList*>(pObj);
      }
      catch (...) {
        error() << "Failed to get DybArchiveList " << objReg.path() << endreq;
        return StatusCode::FAILURE;
      }
      archiveMap[objReg.path()] = pArchiveList;
      //info() << "Indexing archive list at " << objReg.path() << endreq;
    }
  }

  // Define trimming time window
  double window = m_pTrimSvc->window(m_regSeqLocation);
  TimeStamp threshold = pTemporal->earliest();
  threshold.Add(-1.0 * window);
  // Iterate on known paths, and trim  
  std::map<std::string,DybArchiveList*>::iterator archiveMapIter,
    archiveMapEnd=archiveMap.end();
  for(archiveMapIter=archiveMap.begin();
      archiveMapIter!=archiveMapEnd;
      archiveMapIter++){
    //info() << "trimming " << archiveMapIter->first << endreq;
    DybArchiveList* archiveList = archiveMapIter->second;
    if(archiveList->empty()) continue;
    // Stupid hoops I have to jump through to properly reverse iterate
    // and erase on DybArchiveList
    DybArchiveList::iterator archiveIter = archiveList->end();
    DybArchiveList::iterator archiveBegin = archiveList->begin();
    archiveIter--;
    //int count=0;
    while(true){
      DataObject* dataObject = (*archiveIter);
      // Protect against zero object.
      if ( 0 == dataObject) {
        error() << "Broken object in archive! " << archiveMapIter->first 
                << endreq;
        return StatusCode::FAILURE;
      }
      // Cast to Temporal object
      TemporalDataObject* temporalObject = 0;
      try{
        temporalObject = dynamic_cast<TemporalDataObject*>(dataObject);
      }catch(...){
        error() << "Broken temporal object in archive! " 
                << archiveMapIter->first << endreq;
        return StatusCode::FAILURE;
      }
      //info() << "  Checking object: " << archiveMapIter->first 
      //     << " : " << count << endreq;
      if (temporalObject->earliest() <= threshold) {
        //info() << "    Trimming object: " << archiveMapIter->first << endreq;
        static IIncidentSvc* incSvc=0;
        if(!incSvc){
          StatusCode status = service("IncidentSvc",incSvc,true);
          if (status.isFailure()) {
            return status;
          }
        }
        incSvc->fireIncident(EventStoreIncident(temporalObject,false,
                                                "TrimmingAgent"));
        DybArchiveList::iterator tmpIter(archiveIter);
        // Be sure to iterate before deleting target
        bool lastEntry=false;
        if(archiveIter==archiveBegin){
          lastEntry=true;
        }else{
          archiveIter--;
        }
        archiveList->erase(tmpIter);
        if(lastEntry) break;
        //count++;
      }else{
        // This archive is trimmed enough for now
        break;
      }
    }
  }

  return StatusCode::SUCCESS;
}


// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * 

StatusCode DybTrimIO::finalize() 
{
  StatusCode sc;
  debug() << "DybTrimIO finalize()" << endreq;

  // Trim the rest of elements off
  // Some data might still in AES and hasn't been written out.
  // Use the limit of TimeStamp to trigger the saving and trimming
  
  // save before trim
  if(m_saveFlag) {
    sc = m_pSaveB4TrimAesSvc->store(m_regSeqLocation,
                                    m_pDybStorageSvc,
                                    m_pArchiveSvc,
                                    m_pTrimSvc,
                                    TimeStamp::GetEOT());
    if(sc.isFailure()) {
      debug()<<"Failed to write"<<endreq;
      return StatusCode::FAILURE;
    }
  }

  // trim base on it
  sc = m_pTrimSvc->trim(m_pArchiveSvc,TimeStamp::GetEOT());
  if(sc.isFailure()) {
    debug()<<"Failed to trim"<<endreq;
    return StatusCode::FAILURE;
  }

  // Now release these service
  if (0 != m_pTrimSvc) {
    m_pTrimSvc->release();
  }

  if (0 != m_pArchiveSvc) {
    m_pArchiveSvc->release();
  }    

  if(m_saveFlag) {
    if (0 != m_pSaveB4TrimAesSvc) {
      m_pSaveB4TrimAesSvc->release();
    }

    if (0 != m_pDybStorageSvc) {
      m_pDybStorageSvc->release();
    }
  }

  return StatusCode::SUCCESS;
}


// Local Variables: **
// c-basic-offset:2 **
// End: **