#include <DQMArchiver.h>

Public Member Functions
	DQMArchiver (StateMachine *)
const stor::DQMEventConsRegPtr &	getRegPtr () const
	~DQMArchiver ()
Private Types
typedef std::map< std::string, stor::DQMTopLevelFolder::Record >	Records
Private Member Functions
void	activity ()
void	createRegistration ()
void	doIt ()
void	handleDQMEvent (const stor::DQMTopLevelFolder::Record &)
void	updateLastRecord (const stor::DQMTopLevelFolder::Record &)
void	writeDQMEventToFile (const DQMEventMsgView &, const bool endRun) const
Private Attributes
const DQMArchivingParams	dqmArchivingParams_
stor::DQMEventQueueCollectionPtr	dqmEventQueueCollection_
Records	lastUpdateForFolders_
stor::DQMEventConsRegPtr	regPtr_
StateMachine *	stateMachine_
boost::scoped_ptr< boost::thread >	thread_

Detailed Description

Archive DQM histograms

Author:: mommsen

Revision:: 1.2

Date:: 2011/03/07 15:41:54

Definition at line 30 of file DQMArchiver.h.

Member Typedef Documentation

typedef std::map<std::string,stor::DQMTopLevelFolder::Record> smproxy::DQMArchiver::Records [private]

Definition at line 61 of file DQMArchiver.h.

Constructor & Destructor Documentation

smproxy::DQMArchiver::DQMArchiver ( StateMachine * stateMachine )

Definition at line 23 of file DQMArchiver.cc.

References activity(), smproxy::DQMArchivingParams::archiveDQM_, createRegistration(), dqmArchivingParams_, and thread_.

                                                     :
  stateMachine_(stateMachine),
  dqmArchivingParams_(stateMachine->getConfiguration()->getDQMArchivingParams()),
  dqmEventQueueCollection_(stateMachine->getDQMEventQueueCollection())
  {
    if ( dqmArchivingParams_.archiveDQM_ )
    {
      createRegistration();
      thread_.reset(
        new boost::thread( boost::bind(&DQMArchiver::activity, this) )
      );
    }
  }

smproxy::DQMArchiver::~DQMArchiver ( )

Definition at line 37 of file DQMArchiver.cc.

References thread_.

  {
    if (thread_) thread_->join();
  }

Member Function Documentation

void smproxy::DQMArchiver::activity ( ) [private]

Definition at line 42 of file DQMArchiver.cc.

References doIt(), exception, Exception, smproxy::StateMachine::moveToFailedState(), and stateMachine_.

Referenced by DQMArchiver().

  {
    try
    {
      doIt();
    }
    catch(xcept::Exception &e)
    {
      stateMachine_->moveToFailedState(e);
    }
    catch(std::exception &e)
    {
      XCEPT_DECLARE(exception::DQMArchival,
        sentinelException, e.what());
      stateMachine_->moveToFailedState(sentinelException);
    }
    catch(...)
    {
      std::string errorMsg = "Unknown exception in DQM archiving thread";
      XCEPT_DECLARE(exception::DQMArchival,
        sentinelException, errorMsg);
      stateMachine_->moveToFailedState(sentinelException);
    }
  }

void smproxy::DQMArchiver::createRegistration ( ) [private]

Definition at line 159 of file DQMArchiver.cc.

References edm::ParameterSet::addUntrackedParameter(), smproxy::DQMArchivingParams::archiveTopLevelFolder_, dqmArchivingParams_, dqmEventQueueCollection_, smproxy::StateMachine::getConfiguration(), smproxy::StateMachine::getRegistrationCollection(), MultipleCompare::pset, regPtr_, and stateMachine_.

Referenced by DQMArchiver().

  {
    edm::ParameterSet pset;
    pset.addUntrackedParameter<std::string>("DQMconsumerName", "DQMArchiver");
    pset.addUntrackedParameter<std::string>("topLevelFolderName", 
      dqmArchivingParams_.archiveTopLevelFolder_);

    regPtr_.reset( new stor::DQMEventConsumerRegistrationInfo(pset,
        stateMachine_->getConfiguration()->getEventServingParams(),
        "internal")
    );
    
    stor::RegistrationCollectionPtr registrationCollection =
      stateMachine_->getRegistrationCollection();
    
    const stor::ConsumerID cid = registrationCollection->getConsumerId();
    regPtr_->setConsumerId(cid);

    const stor::QueueID qid = dqmEventQueueCollection_->createQueue(regPtr_);
    regPtr_->setQueueId(qid);

    registrationCollection->addRegistrationInfo(regPtr_);
  }

void smproxy::DQMArchiver::doIt ( ) [private]

Definition at line 67 of file DQMArchiver.cc.

References dqmEventQueueCollection_, smproxy::StateMachine::getRegistrationCollection(), handleDQMEvent(), lastUpdateForFolders_, regPtr_, stor::utils::sleep(), stateMachine_, and writeDQMEventToFile().

Referenced by activity().

  {
    stor::RegistrationCollectionPtr registrationCollection =
      stateMachine_->getRegistrationCollection();
    const stor::ConsumerID cid = regPtr_->consumerId();
    
    while ( registrationCollection->registrationIsAllowed(cid) )
    {
      const stor::DQMEventQueueCollection::ValueType dqmEvent =
        dqmEventQueueCollection_->popEvent(cid);
      
      if ( dqmEvent.first.empty() )
        ::sleep(1);
      else
        handleDQMEvent(dqmEvent.first);
    }

    // run ended, write the last updates to file
    BOOST_FOREACH(
      const Records::value_type& pair,
      lastUpdateForFolders_
    )
    {
      writeDQMEventToFile(pair.second.getDQMEventMsgView(), true);
    }
  }

const stor::DQMEventConsRegPtr& smproxy::DQMArchiver::getRegPtr ( ) const [inline]

Return the registration pointer

Definition at line 41 of file DQMArchiver.h.

References regPtr_.

Referenced by smproxy::DataManager::doIt().

    { return regPtr_; }

void smproxy::DQMArchiver::handleDQMEvent ( const stor::DQMTopLevelFolder::Record & record ) [private]

Definition at line 94 of file DQMArchiver.cc.

References smproxy::DQMArchivingParams::archiveIntervalDQM_, dqmArchivingParams_, stor::DQMTopLevelFolder::Record::getDQMEventMsgView(), updateLastRecord(), DQMEventMsgView::updateNumber(), and writeDQMEventToFile().

Referenced by doIt().

  {
    const DQMEventMsgView view = record.getDQMEventMsgView();

    updateLastRecord(record);

    if (
      dqmArchivingParams_.archiveIntervalDQM_ > 0 &&
      ((view.updateNumber()+1) % dqmArchivingParams_.archiveIntervalDQM_) == 0
    )
    {
      writeDQMEventToFile(view, false);
    }
  }

void smproxy::DQMArchiver::updateLastRecord ( const stor::DQMTopLevelFolder::Record & record ) [private]

Definition at line 109 of file DQMArchiver.cc.

References stor::DQMTopLevelFolder::Record::getDQMEventMsgView(), lastUpdateForFolders_, pos, record, and DQMEventMsgView::topFolderName().

Referenced by handleDQMEvent().

  {
    const DQMEventMsgView view = record.getDQMEventMsgView();
    const std::string topFolderName = view.topFolderName();
    Records::iterator pos = lastUpdateForFolders_.lower_bound(topFolderName);

    if (pos != lastUpdateForFolders_.end() &&
      !(lastUpdateForFolders_.key_comp()(topFolderName, pos->first)))
    {
      // key already exists
      pos->second = record;
    }
    else
    {
      lastUpdateForFolders_.insert(pos, Records::value_type(topFolderName, record));
    }
  }

void smproxy::DQMArchiver::writeDQMEventToFile	(	const DQMEventMsgView &	view,
		const bool	endRun
	)		const `[private]`

Definition at line 128 of file DQMArchiver.cc.

References edm::DQMHttpSource::addEventToDQMBackend(), edm::ParameterSet::addUntrackedParameter(), convertXMLtoSQLite_cfg::fileName, stor::DQMEventMonitorCollection::getNumberOfWrittenTopLevelFoldersMQ(), stor::DQMEventMonitorCollection::getWrittenDQMEventSizeMQ(), DQMEventMsgView::lumiSection(), DQMEventMsgView::runNumber(), DQMEventMsgView::size(), and DQMStore::StripRunDirs.

Referenced by doIt(), and handleDQMEvent().

  {
    edm::ParameterSet dqmStorePSet;
    dqmStorePSet.addUntrackedParameter<bool>("collateHistograms", true);
    DQMStore dqmStore(dqmStorePSet);

    std::ostringstream fileName;
    fileName << dqmArchivingParams_.filePrefixDQM_
      << "/DQM_R"
      << std::setfill('0') << std::setw(9) << view.runNumber();
    if ( ! endRun ) fileName << "_L" << std::setw(6) << view.lumiSection();
    fileName << ".root";
    
    // don't require that the file exists
    dqmStore.load(fileName.str(), DQMStore::StripRunDirs, false);

    edm::DQMHttpSource::addEventToDQMBackend(&dqmStore, view, false);

    dqmStore.save(fileName.str());

    stor::DQMEventMonitorCollection& demc =
      stateMachine_->getStatisticsReporter()->getDQMEventMonitorCollection();
    demc.getWrittenDQMEventSizeMQ().addSample(
      static_cast<double>(view.size()) / 0x100000
    );
    demc.getNumberOfWrittenTopLevelFoldersMQ().addSample(1);
  }