stor::StreamsMonitorCollection Class Reference

#include <StreamsMonitorCollection.h>

Inheritance diagram for stor::StreamsMonitorCollection:

Classes
struct	EndOfRunReport
struct	StreamRecord

Public Types
typedef boost::shared_ptr < EndOfRunReport >	EndOfRunReportPtr
typedef std::vector < StreamRecordPtr >	StreamRecordList
typedef boost::shared_ptr < StreamRecord >	StreamRecordPtr
Public Types inherited from stor::MonitorCollection
typedef std::vector< std::pair < std::string, xdata::Serializable * > >	InfoSpaceItems

Public Member Functions
const MonitoredQuantity &	getAllStreamsBandwidthMQ () const
MonitoredQuantity &	getAllStreamsBandwidthMQ ()
const MonitoredQuantity &	getAllStreamsFileCountMQ () const
MonitoredQuantity &	getAllStreamsFileCountMQ ()
const MonitoredQuantity &	getAllStreamsVolumeMQ () const
MonitoredQuantity &	getAllStreamsVolumeMQ ()
StreamRecordPtr	getNewStreamRecord ()
void	getStreamRecords (StreamRecordList &) const
bool	getStreamRecordsForOutputModuleLabel (const std::string &, StreamRecordList &) const
void	reportAllLumiSectionInfos (DbFileHandlerPtr, EndOfRunReportPtr)
bool	streamRecordsExist () const
	StreamsMonitorCollection (const utils::Duration_t &updateInterval)
Public Member Functions inherited from stor::MonitorCollection
void	appendInfoSpaceItems (InfoSpaceItems &)
void	calculateStatistics (const utils::TimePoint_t &now)
	MonitorCollection (const utils::Duration_t &updateInterval)
void	reset (const utils::TimePoint_t &now)
void	updateInfoSpaceItems ()
virtual	~MonitorCollection ()

Private Types
typedef std::set< uint32_t >	UnreportedLS

Private Member Functions
virtual void	do_appendInfoSpaceItems (InfoSpaceItems &)
virtual void	do_calculateStatistics ()
virtual void	do_reset ()
virtual void	do_updateInfoSpaceItems ()
void	getListOfAllUnreportedLS (UnreportedLS &)
StreamsMonitorCollection &	operator= (StreamsMonitorCollection const &)
	StreamsMonitorCollection (StreamsMonitorCollection const &)

Private Attributes
MonitoredQuantity	allStreamsBandwidth_
MonitoredQuantity	allStreamsFileCount_
MonitoredQuantity	allStreamsVolume_
xdata::Vector< xdata::Double >	bandwidthPerStream_
xdata::Double	bandwidthToDisk_
xdata::Vector < xdata::UnsignedInteger32 >	eventsPerStream_
xdata::Vector< xdata::Double >	ratePerStream_
xdata::UnsignedInteger32	storedEvents_
xdata::Double	storedVolume_
xdata::Vector< xdata::String >	streamNames_
StreamRecordList	streamRecords_
boost::mutex	streamRecordsMutex_
const utils::Duration_t	timeWindowForRecentResults_
const utils::Duration_t	updateInterval_

Additional Inherited Members

Detailed Description

A collection of MonitoredQuantities of output streams

Author:

mommsen

Revision:

1.17

Date:

2011/11/17 17:35:41

Definition at line 35 of file StreamsMonitorCollection.h.

Member Typedef Documentation

typedef boost::shared_ptr<EndOfRunReport> stor::StreamsMonitorCollection::EndOfRunReportPtr

Definition at line 101 of file StreamsMonitorCollection.h.

typedef std::vector<StreamRecordPtr> stor::StreamsMonitorCollection::StreamRecordList

Definition at line 82 of file StreamsMonitorCollection.h.

typedef boost::shared_ptr<StreamRecord> stor::StreamsMonitorCollection::StreamRecordPtr

Definition at line 81 of file StreamsMonitorCollection.h.

typedef std::set<uint32_t> stor::StreamsMonitorCollection::UnreportedLS

private

Definition at line 144 of file StreamsMonitorCollection.h.

Constructor & Destructor Documentation

stor::StreamsMonitorCollection::StreamsMonitorCollection ( const utils::Duration_t &  updateInterval )

explicit

Definition at line 15 of file StreamsMonitorCollection.cc.

    :
  MonitorCollection(updateInterval),
  updateInterval_(updateInterval),
  timeWindowForRecentResults_(boost::posix_time::seconds(10)),
  allStreamsFileCount_(updateInterval, timeWindowForRecentResults_),
  allStreamsVolume_(updateInterval, timeWindowForRecentResults_),
  allStreamsBandwidth_(updateInterval, timeWindowForRecentResults_)
  {}

stor::StreamsMonitorCollection::StreamsMonitorCollection ( StreamsMonitorCollection const &  )

private

Member Function Documentation

void stor::StreamsMonitorCollection::do_appendInfoSpaceItems ( InfoSpaceItems &  infoSpaceItems )

privatevirtual

Reimplemented from stor::MonitorCollection.

Definition at line 228 of file StreamsMonitorCollection.cc.

References bandwidthPerStream_, bandwidthToDisk_, eventsPerStream_, ratePerStream_, storedEvents_, storedVolume_, and streamNames_.

  {
    infoSpaceItems.push_back(std::make_pair("storedEvents",  &storedEvents_));
    infoSpaceItems.push_back(std::make_pair("storedVolume",  &storedVolume_));
    infoSpaceItems.push_back(std::make_pair("bandwidthToDisk",  &bandwidthToDisk_));
    infoSpaceItems.push_back(std::make_pair("streamNames",  &streamNames_));
    infoSpaceItems.push_back(std::make_pair("eventsPerStream",  &eventsPerStream_));
    infoSpaceItems.push_back(std::make_pair("ratePerStream",  &ratePerStream_));
    infoSpaceItems.push_back(std::make_pair("bandwidthPerStream",  &bandwidthPerStream_));
  }

void stor::StreamsMonitorCollection::do_calculateStatistics ( )

privatevirtual

Implements stor::MonitorCollection.

Definition at line 193 of file StreamsMonitorCollection.cc.

References stor::MonitoredQuantity::addSample(), allStreamsBandwidth_, allStreamsFileCount_, allStreamsVolume_, stor::MonitoredQuantity::calculateStatistics(), stor::MonitoredQuantity::Stats::getLastValueRate(), stor::MonitoredQuantity::Stats::getSampleCount(), stor::MonitoredQuantity::getStats(), streamRecords_, and streamRecordsMutex_.

  {
    MonitoredQuantity::Stats stats;
    
    allStreamsFileCount_.calculateStatistics();
    allStreamsVolume_.calculateStatistics();
    allStreamsVolume_.getStats(stats);
    bool samplingHasStarted = (stats.getSampleCount() > 0);
    if (samplingHasStarted) {
      allStreamsBandwidth_.addSample(stats.getLastValueRate());
    }
    allStreamsBandwidth_.calculateStatistics();
    
    
    boost::mutex::scoped_lock sl(streamRecordsMutex_);
    
    for (
      StreamRecordList::const_iterator 
        it = streamRecords_.begin(), itEnd = streamRecords_.end();
      it != itEnd;
      ++it
    ) 
    {
      (*it)->fileCount.calculateStatistics();
      (*it)->volume.calculateStatistics();
      (*it)->volume.getStats(stats);
      if (samplingHasStarted) {
        (*it)->bandwidth.addSample(stats.getLastValueRate());
      }
      (*it)->bandwidth.calculateStatistics();
    }
  }

void stor::StreamsMonitorCollection::do_reset ( )

privatevirtual

Implements stor::MonitorCollection.

Definition at line 242 of file StreamsMonitorCollection.cc.

References allStreamsBandwidth_, allStreamsFileCount_, allStreamsVolume_, stor::MonitoredQuantity::reset(), streamRecords_, and streamRecordsMutex_.

  {
    allStreamsFileCount_.reset();
    allStreamsVolume_.reset();
    allStreamsBandwidth_.reset();
    
    boost::mutex::scoped_lock sl(streamRecordsMutex_);
    streamRecords_.clear();
  }

void stor::StreamsMonitorCollection::do_updateInfoSpaceItems ( )

privatevirtual

Reimplemented from stor::MonitorCollection.

Definition at line 253 of file StreamsMonitorCollection.cc.

References allStreamsVolume_, bandwidthPerStream_, bandwidthToDisk_, eventsPerStream_, stor::MonitoredQuantity::FULL, stor::MonitoredQuantity::Stats::getSampleCount(), stor::MonitoredQuantity::Stats::getSampleRate(), stor::MonitoredQuantity::getStats(), stor::MonitoredQuantity::Stats::getValueRate(), stor::MonitoredQuantity::Stats::getValueSum(), ratePerStream_, stor::MonitoredQuantity::RECENT, storedEvents_, storedVolume_, streamNames_, streamRecords_, and streamRecordsMutex_.

  {
    MonitoredQuantity::Stats allStreamsVolumeStats;
    allStreamsVolume_.getStats(allStreamsVolumeStats);
    
    storedEvents_ = static_cast<xdata::UnsignedInteger32>(
      allStreamsVolumeStats.getSampleCount()
    );
    storedVolume_ = static_cast<xdata::Double>(
      allStreamsVolumeStats.getValueSum()
    );
    bandwidthToDisk_ = static_cast<xdata::Double>(
      allStreamsVolumeStats.getValueRate(MonitoredQuantity::RECENT)
    );
    
    boost::mutex::scoped_lock sl(streamRecordsMutex_);
    
    streamNames_.clear();
    eventsPerStream_.clear();
    ratePerStream_.clear();
    bandwidthPerStream_.clear();
    
    streamNames_.reserve(streamRecords_.size());
    eventsPerStream_.reserve(streamRecords_.size());
    ratePerStream_.reserve(streamRecords_.size());
    bandwidthPerStream_.reserve(streamRecords_.size());
    
    for (
      StreamRecordList::const_iterator
        it = streamRecords_.begin(), itEnd = streamRecords_.end();
      it != itEnd;
      ++it
    )
    {
      MonitoredQuantity::Stats streamVolumeStats;
      (*it)->volume.getStats(streamVolumeStats);
      MonitoredQuantity::Stats streamBandwidthStats;
      (*it)->bandwidth.getStats(streamBandwidthStats);
      
      streamNames_.push_back(
        static_cast<xdata::String>( (*it)->streamName )
      );
      
      eventsPerStream_.push_back(
        static_cast<xdata::UnsignedInteger32>(
          streamVolumeStats.getSampleCount(MonitoredQuantity::FULL)
        )
      );
      
      ratePerStream_.push_back(
        static_cast<xdata::Double>(
          streamVolumeStats.getSampleRate(MonitoredQuantity::RECENT)
        )
      );
      
      bandwidthPerStream_.push_back(
        static_cast<xdata::Double>(
          streamBandwidthStats.getValueRate(MonitoredQuantity::RECENT)
        )
      );
    }
  }

const MonitoredQuantity& stor::StreamsMonitorCollection::getAllStreamsBandwidthMQ ( ) const

inline

Definition at line 128 of file StreamsMonitorCollection.h.

References allStreamsBandwidth_.

Referenced by stor::SMWebPageHelper::listStreamRecordsStats().

                                                              {
      return allStreamsBandwidth_;
    }

MonitoredQuantity& stor::StreamsMonitorCollection::getAllStreamsBandwidthMQ ( )

inline

Definition at line 131 of file StreamsMonitorCollection.h.

References allStreamsBandwidth_.

                                                  {
      return allStreamsBandwidth_;
    }

const MonitoredQuantity& stor::StreamsMonitorCollection::getAllStreamsFileCountMQ ( ) const

inline

Definition at line 114 of file StreamsMonitorCollection.h.

References allStreamsFileCount_.

Referenced by stor::SMWebPageHelper::listStreamRecordsStats().

                                                              {
      return allStreamsFileCount_;
    }

MonitoredQuantity& stor::StreamsMonitorCollection::getAllStreamsFileCountMQ ( )

inline

Definition at line 117 of file StreamsMonitorCollection.h.

References allStreamsFileCount_.

                                                  {
      return allStreamsFileCount_;
    }

const MonitoredQuantity& stor::StreamsMonitorCollection::getAllStreamsVolumeMQ ( ) const

inline

Definition at line 121 of file StreamsMonitorCollection.h.

References allStreamsVolume_.

Referenced by stor::SMWebPageHelper::addDOMforStoredData(), and stor::SMWebPageHelper::listStreamRecordsStats().

                                                           {
      return allStreamsVolume_;
    }

MonitoredQuantity& stor::StreamsMonitorCollection::getAllStreamsVolumeMQ ( )

inline

Definition at line 124 of file StreamsMonitorCollection.h.

References allStreamsVolume_.

                                               {
      return allStreamsVolume_;
    }

void stor::StreamsMonitorCollection::getListOfAllUnreportedLS ( UnreportedLS &  unreportedLS )

private

Definition at line 172 of file StreamsMonitorCollection.cc.

References streamRecords_.

Referenced by reportAllLumiSectionInfos().

  {
    // Have to loop over all streams as not every stream
    // might have got an event for a given lumi section
    for (StreamRecordList::const_iterator 
           stream = streamRecords_.begin(),
           streamEnd = streamRecords_.end();
         stream != streamEnd;
         ++stream)
    {
      for (StreamRecord::FileCountPerLumiSectionMap::const_iterator
             lscount = (*stream)->fileCountPerLS.begin(),
             lscountEnd = (*stream)->fileCountPerLS.end();
           lscount != lscountEnd; ++lscount)
      {
        unreportedLS.insert(lscount->first);
      }
    }
  }

StreamsMonitorCollection::StreamRecordPtr stor::StreamsMonitorCollection::getNewStreamRecord

(

)

Definition at line 28 of file StreamsMonitorCollection.cc.

References streamRecords_, streamRecordsMutex_, timeWindowForRecentResults_, and updateInterval_.

  {
    boost::mutex::scoped_lock sl(streamRecordsMutex_);
    
    StreamRecordPtr streamRecord(
      new StreamRecord(this,updateInterval_,timeWindowForRecentResults_)
    );
    streamRecords_.push_back(streamRecord);
    return streamRecord;
  }

void stor::StreamsMonitorCollection::getStreamRecords

(

StreamRecordList &

list

)

const

Definition at line 40 of file StreamsMonitorCollection.cc.

References streamRecords_, and streamRecordsMutex_.

Referenced by stor::SMWebPageHelper::addDOMforSummaryInformation(), and stor::SMWebPageHelper::listStreamRecordsStats().

  {
    boost::mutex::scoped_lock sl(streamRecordsMutex_);

    list.clear();
    list.reserve(streamRecords_.size());
    
    for (
      StreamRecordList::const_iterator 
        it = streamRecords_.begin(), itEnd = streamRecords_.end();
      it != itEnd;
      ++it
    )
    {
      list.push_back(*it);
    }
  }

bool stor::StreamsMonitorCollection::getStreamRecordsForOutputModuleLabel	(	const std::string &	label,
		StreamRecordList &	list
	)		const

Definition at line 60 of file StreamsMonitorCollection.cc.

Referenced by stor::SMWebPageHelper::addDOMforSummaryInformation().

  {
    boost::mutex::scoped_lock sl(streamRecordsMutex_);

    list.clear();
    list.reserve(streamRecords_.size());
    
    for (
      StreamRecordList::const_iterator 
        it = streamRecords_.begin(), itEnd = streamRecords_.end();
      it != itEnd;
      ++it
    )
    {
      if ( (*it)->outputModuleLabel == label )
        list.push_back(*it);
    }
    return ( ! list.empty() );
  }

StreamsMonitorCollection& stor::StreamsMonitorCollection::operator= ( StreamsMonitorCollection const &  )

private

void stor::StreamsMonitorCollection::reportAllLumiSectionInfos	(	DbFileHandlerPtr	dbFileHandler,
		EndOfRunReportPtr	endOfRunReport
	)

Definition at line 138 of file StreamsMonitorCollection.cc.

References getListOfAllUnreportedLS(), streamRecords_, and streamRecordsMutex_.

Referenced by stor::DiskWriter::reportRemainingLumiSections().

  {
    boost::mutex::scoped_lock sl(streamRecordsMutex_);

    UnreportedLS unreportedLS;
    getListOfAllUnreportedLS(unreportedLS);
    
    for (UnreportedLS::const_iterator it = unreportedLS.begin(),
           itEnd = unreportedLS.end(); it != itEnd; ++it)
    {
      std::string lsEntry;
      bool filesWritten = false;

      for (StreamRecordList::const_iterator 
             stream = streamRecords_.begin(),
             streamEnd = streamRecords_.end();
           stream != streamEnd;
           ++stream)
      {
        if ( (*stream)->reportLumiSectionInfo((*it), lsEntry) )
          filesWritten = true;
      }
      lsEntry += "\tEoLS:0";
      dbFileHandler->write(lsEntry);

      if (filesWritten) ++(endOfRunReport->lsCountWithFiles);
      endOfRunReport->updateLatestWrittenLumiSection(*it);
    }
  }

bool stor::StreamsMonitorCollection::streamRecordsExist

(

)

const

Definition at line 84 of file StreamsMonitorCollection.cc.

References streamRecords_, and streamRecordsMutex_.

Referenced by stor::SMWebPageHelper::addDOMforStoredData().

  {
    boost::mutex::scoped_lock sl(streamRecordsMutex_);

    return ( ! streamRecords_.empty() );
  }

Member Data Documentation

MonitoredQuantity stor::StreamsMonitorCollection::allStreamsBandwidth_

private

Definition at line 160 of file StreamsMonitorCollection.h.

Referenced by do_calculateStatistics(), do_reset(), and getAllStreamsBandwidthMQ().

MonitoredQuantity stor::StreamsMonitorCollection::allStreamsFileCount_

private

Definition at line 158 of file StreamsMonitorCollection.h.

Referenced by do_calculateStatistics(), do_reset(), and getAllStreamsFileCountMQ().

MonitoredQuantity stor::StreamsMonitorCollection::allStreamsVolume_

private

Definition at line 159 of file StreamsMonitorCollection.h.

Referenced by stor::StreamsMonitorCollection::StreamRecord::addSizeInBytes(), do_calculateStatistics(), do_reset(), do_updateInfoSpaceItems(), and getAllStreamsVolumeMQ().

xdata::Vector<xdata::Double> stor::StreamsMonitorCollection::bandwidthPerStream_

private

Definition at line 168 of file StreamsMonitorCollection.h.

Referenced by do_appendInfoSpaceItems(), and do_updateInfoSpaceItems().

xdata::Double stor::StreamsMonitorCollection::bandwidthToDisk_

private

Definition at line 164 of file StreamsMonitorCollection.h.

Referenced by do_appendInfoSpaceItems(), and do_updateInfoSpaceItems().

xdata::Vector<xdata::UnsignedInteger32> stor::StreamsMonitorCollection::eventsPerStream_

private

Definition at line 166 of file StreamsMonitorCollection.h.

Referenced by do_appendInfoSpaceItems(), and do_updateInfoSpaceItems().

xdata::Vector<xdata::Double> stor::StreamsMonitorCollection::ratePerStream_

private

Definition at line 167 of file StreamsMonitorCollection.h.

Referenced by do_appendInfoSpaceItems(), and do_updateInfoSpaceItems().

xdata::UnsignedInteger32 stor::StreamsMonitorCollection::storedEvents_

private

Definition at line 162 of file StreamsMonitorCollection.h.

Referenced by do_appendInfoSpaceItems(), and do_updateInfoSpaceItems().

xdata::Double stor::StreamsMonitorCollection::storedVolume_

private

Definition at line 163 of file StreamsMonitorCollection.h.

Referenced by do_appendInfoSpaceItems(), and do_updateInfoSpaceItems().

xdata::Vector<xdata::String> stor::StreamsMonitorCollection::streamNames_

private

Definition at line 165 of file StreamsMonitorCollection.h.

Referenced by do_appendInfoSpaceItems(), and do_updateInfoSpaceItems().

StreamRecordList stor::StreamsMonitorCollection::streamRecords_

private

Definition at line 152 of file StreamsMonitorCollection.h.

Referenced by do_calculateStatistics(), do_reset(), do_updateInfoSpaceItems(), getListOfAllUnreportedLS(), getNewStreamRecord(), getStreamRecords(), reportAllLumiSectionInfos(), and streamRecordsExist().

boost::mutex stor::StreamsMonitorCollection::streamRecordsMutex_

mutableprivate

Definition at line 153 of file StreamsMonitorCollection.h.

Referenced by do_calculateStatistics(), do_reset(), do_updateInfoSpaceItems(), getNewStreamRecord(), getStreamRecords(), reportAllLumiSectionInfos(), and streamRecordsExist().

const utils::Duration_t stor::StreamsMonitorCollection::timeWindowForRecentResults_

private

Definition at line 156 of file StreamsMonitorCollection.h.

Referenced by getNewStreamRecord().

const utils::Duration_t stor::StreamsMonitorCollection::updateInterval_

private

Definition at line 155 of file StreamsMonitorCollection.h.

Referenced by getNewStreamRecord().