#include <MonitoredQuantity.h>

Classes
struct	Stats
Public Types
enum	DataSetType { FULL = 0, RECENT = 1 }
Public Member Functions
void	addSample (const double &value=1)
void	addSample (const unsigned int &value=1)
void	addSample (const unsigned long long &value=1)
void	addSample (const long &value=1)
void	addSample (const int &value=1)
void	addSample (const unsigned long &value=1)
void	addSample (const long long &value=1)
void	addSampleIfLarger (const double &value)
void	calculateStatistics (const utils::TimePoint_t &currentTime=utils::getCurrentTime())
void	disable ()
void	enable ()
utils::Duration_t	ExpectedCalculationInterval () const
void	getStats (Stats &stats) const
utils::Duration_t	getTimeWindowForRecentResults () const
bool	isEnabled () const
	MonitoredQuantity (utils::Duration_t expectedCalculationInterval, utils::Duration_t timeWindowForRecentResults)
void	reset ()
void	setNewTimeWindowForRecentResults (const utils::Duration_t &interval)
Private Member Functions
	MonitoredQuantity (MonitoredQuantity const &)
MonitoredQuantity &	operator= (MonitoredQuantity const &)
void	resetAccumulators ()
void	resetResults ()
Private Attributes
boost::mutex	accumulationMutex_
uint32_t	binCount_
std::vector< utils::Duration_t >	binDuration_
std::vector< uint64_t >	binSampleCount_
std::vector< utils::TimePoint_t >	binSnapshotTime_
std::vector< double >	binValueMax_
std::vector< double >	binValueMin_
std::vector< double >	binValueSum_
std::vector< double >	binValueSumOfSquares_
bool	enabled_
const utils::Duration_t	expectedCalculationInterval_
utils::Duration_t	fullDuration_
uint64_t	fullSampleCount_
double	fullSampleRate_
double	fullValueAverage_
double	fullValueMax_
double	fullValueMin_
double	fullValueRate_
double	fullValueRMS_
double	fullValueSum_
double	fullValueSumOfSquares_
utils::Duration_t	intervalForRecentStats_
utils::TimePoint_t	lastCalculationTime_
double	lastLatchedSampleValue_
double	lastLatchedValueRate_
utils::Duration_t	recentDuration_
uint64_t	recentSampleCount_
double	recentSampleRate_
double	recentValueAverage_
double	recentValueMax_
double	recentValueMin_
double	recentValueRate_
double	recentValueRMS_
double	recentValueSum_
double	recentValueSumOfSquares_
boost::mutex	resultsMutex_
uint32_t	workingBinId_
double	workingLastSampleValue_
uint64_t	workingSampleCount_
double	workingValueMax_
double	workingValueMin_
double	workingValueSum_
double	workingValueSumOfSquares_

Detailed Description

This class keeps track of statistics for a set of sample values and provides timing information on the samples.

Author:: mommsen

Revision:: 1.10

Date:: 2011/03/07 15:31:32

Definition at line 29 of file MonitoredQuantity.h.

Member Enumeration Documentation

enum stor::MonitoredQuantity::DataSetType

Enumerator:

FULL
RECENT

Definition at line 35 of file MonitoredQuantity.h.

                     { FULL = 0,      // the full data set (all samples)
                       RECENT = 1 };  // recent data only

Constructor & Destructor Documentation

stor::MonitoredQuantity::MonitoredQuantity	(	utils::Duration_t	expectedCalculationInterval,
		utils::Duration_t	timeWindowForRecentResults
	)		`[explicit]`

Definition at line 13 of file MonitoredQuantity.cc.

   :
  enabled_(true),
  expectedCalculationInterval_(expectedCalculationInterval)
  {
    setNewTimeWindowForRecentResults(timeWindowForRecentResults);
  }

stor::MonitoredQuantity::MonitoredQuantity ( MonitoredQuantity const & ) [private]

Member Function Documentation

void stor::MonitoredQuantity::addSample ( const double & value = 1 )

Adds the specified doubled valued sample value to the monitor instance.

Definition at line 23 of file MonitoredQuantity.cc.

References accumulationMutex_, enabled_, stor::utils::getCurrentTime(), lastCalculationTime_, relativeConstraints::value, workingLastSampleValue_, workingSampleCount_, workingValueMax_, workingValueMin_, workingValueSum_, and workingValueSumOfSquares_.

  {
    if (! enabled_) {return;}
    
    boost::mutex::scoped_lock sl(accumulationMutex_);
    
    if ( lastCalculationTime_.is_not_a_date_time() )
    {
      lastCalculationTime_ = utils::getCurrentTime();
    }
    
    ++workingSampleCount_;
    workingValueSum_ += value;
    workingValueSumOfSquares_ += (value * value);
    
    if (value < workingValueMin_) workingValueMin_ = value;
    if (value > workingValueMax_) workingValueMax_ = value;
    
    workingLastSampleValue_ = value;
  }

void stor::MonitoredQuantity::addSample ( const int & value = 1 )

Adds the specified integer valued sample value to the monitor instance.

Definition at line 44 of file MonitoredQuantity.cc.

References addSample().

  {
    addSample(static_cast<double>(value));
  }

void stor::MonitoredQuantity::addSample ( const long & value = 1 )

Adds the specified long valued sample value to the monitor instance.

Definition at line 54 of file MonitoredQuantity.cc.

References addSample().

  {
    addSample(static_cast<double>(value));
  }

void stor::MonitoredQuantity::addSample ( const unsigned long & value = 1 )

Adds the specified unsigned long valued sample value to the monitor instance.

Definition at line 59 of file MonitoredQuantity.cc.

References addSample().

  {
    addSample(static_cast<double>(value));
  }

void stor::MonitoredQuantity::addSample ( const unsigned int & value = 1 )

Adds the specified unsigned integer valued sample value to the monitor instance.

Definition at line 49 of file MonitoredQuantity.cc.

References addSample().

  {
    addSample(static_cast<double>(value));
  }

void stor::MonitoredQuantity::addSample ( const long long & value = 1 )

Adds the specified long long valued sample value to the monitor instance.

Definition at line 64 of file MonitoredQuantity.cc.

References addSample().

  {
    addSample(static_cast<double>(value));
  }

void stor::MonitoredQuantity::addSample ( const unsigned long long & value = 1 )

Adds the specified unsigned long long valued sample value to the monitor instance.

Definition at line 69 of file MonitoredQuantity.cc.

References addSample().

  {
    addSample(static_cast<double>(value));
  }

void stor::MonitoredQuantity::addSampleIfLarger ( const double & value )

Adds the specified double valued sample value to the monitor instance if it is larger than the previously added value.

Definition at line 74 of file MonitoredQuantity.cc.

References addSample(), and workingLastSampleValue_.

Referenced by stor::EventDistributor::tagCompleteEventForQueues().

  {
    if (value > workingLastSampleValue_)
      addSample(value);
  }

void stor::MonitoredQuantity::calculateStatistics ( const utils::TimePoint_t & currentTime = utils::getCurrentTime() )

Forces a calculation of the statistics for the monitored quantity. The frequency of the updates to the statistics is driven by how often this method is called. It is expected that this method will be called once per interval specified by expectedCalculationInterval

Definition at line 80 of file MonitoredQuantity.cc.

References accumulationMutex_, binCount_, binDuration_, binSampleCount_, binSnapshotTime_, binValueMax_, binValueMin_, binValueSum_, binValueSumOfSquares_, stor::utils::durationToSeconds(), enabled_, expectedCalculationInterval_, fullDuration_, fullSampleCount_, fullSampleRate_, fullValueAverage_, fullValueMax_, fullValueMin_, fullValueRate_, fullValueRMS_, fullValueSum_, fullValueSumOfSquares_, UserOptions_cff::idx, lastCalculationTime_, lastLatchedSampleValue_, lastLatchedValueRate_, recentDuration_, recentSampleCount_, recentSampleRate_, recentValueAverage_, recentValueMax_, recentValueMin_, recentValueRate_, recentValueRMS_, recentValueSum_, recentValueSumOfSquares_, resultsMutex_, seconds(), mathSSE::sqrt(), workingBinId_, workingLastSampleValue_, workingSampleCount_, workingValueMax_, workingValueMin_, workingValueSum_, and workingValueSumOfSquares_.

Referenced by stor::ThroughputMonitorCollection::calcPoolUsage(), stor::DQMEventMonitorCollection::do_calculateStatistics(), stor::ThroughputMonitorCollection::do_calculateStatistics(), stor::StreamsMonitorCollection::do_calculateStatistics(), stor::RunMonitorCollection::do_calculateStatistics(), stor::ConsumerMonitorCollection::do_calculateStatistics(), and stor::FragmentMonitorCollection::do_calculateStatistics().

  {
    if (! enabled_) {return;}
    
    // create local copies of the working values to minimize the
    // time that we could block a thread trying to add a sample.
    // Also, reset the working values.
    long long latestSampleCount;
    double latestValueSum;
    double latestValueSumOfSquares;
    double latestValueMin;
    double latestValueMax;
    utils::Duration_t latestDuration;
    utils::TimePoint_t latestSnapshotTime;
    double latestLastLatchedSampleValue;
    {
      boost::mutex::scoped_lock sl(accumulationMutex_);

      if (lastCalculationTime_.is_not_a_date_time()) {return;}
      if (currentTime - lastCalculationTime_ < expectedCalculationInterval_) {return;}
      
      latestSampleCount = workingSampleCount_;
      latestValueSum = workingValueSum_;
      latestValueSumOfSquares = workingValueSumOfSquares_;
      latestValueMin = workingValueMin_;
      latestValueMax = workingValueMax_;
      latestDuration = currentTime - lastCalculationTime_;
      latestSnapshotTime = currentTime;
      latestLastLatchedSampleValue = workingLastSampleValue_;
      
      lastCalculationTime_ = currentTime;
      workingSampleCount_ = 0;
      workingValueSum_ = 0.0;
      workingValueSumOfSquares_ = 0.0;
      workingValueMin_ =  INFINITY;
      workingValueMax_ = -INFINITY;
    }
    
    // lock out any interaction with the results while we update them
    {
      boost::mutex::scoped_lock sl(resultsMutex_);
      lastLatchedSampleValue_ = latestLastLatchedSampleValue;
      
      // we simply add the latest results to the full set
      fullSampleCount_ += latestSampleCount;
      fullValueSum_ += latestValueSum;
      fullValueSumOfSquares_ += latestValueSumOfSquares;
      if (latestValueMin < fullValueMin_) {fullValueMin_ = latestValueMin;}
      if (latestValueMax > fullValueMax_) {fullValueMax_ = latestValueMax;}
      fullDuration_ += latestDuration;
      
      // for the recent results, we need to replace the contents of
      // the working bin and re-calculate the recent values
      binSampleCount_[workingBinId_] = latestSampleCount;
      binValueSum_[workingBinId_] = latestValueSum;
      binValueSumOfSquares_[workingBinId_] = latestValueSumOfSquares;
      binValueMin_[workingBinId_] = latestValueMin;
      binValueMax_[workingBinId_] = latestValueMax;
      binDuration_[workingBinId_] = latestDuration;
      binSnapshotTime_[workingBinId_] = latestSnapshotTime;
      
      lastLatchedValueRate_ = latestValueSum / utils::durationToSeconds(latestDuration);
      
      recentSampleCount_ = 0;
      recentValueSum_ = 0.0;
      recentValueSumOfSquares_ = 0.0;
      recentValueMin_ =  INFINITY;
      recentValueMax_ = -INFINITY;
      recentDuration_ = boost::posix_time::seconds(0);
      
      for (unsigned int idx = 0; idx < binCount_; ++idx) {
        recentSampleCount_ += binSampleCount_[idx];
        recentValueSum_ += binValueSum_[idx];
        recentValueSumOfSquares_ += binValueSumOfSquares_[idx];
        if (binValueMin_[idx] < recentValueMin_) {
          recentValueMin_ = binValueMin_[idx];
        }
        if (binValueMax_[idx] > recentValueMax_) {
          recentValueMax_ = binValueMax_[idx];
        }
        recentDuration_ += binDuration_[idx];
      }
      
      // update the working bin ID here so that we are ready for
      // the next calculation request
      ++workingBinId_;
      if (workingBinId_ >= binCount_) {workingBinId_ = 0;}
      
      // calculate the derived full values
      const double fullDuration = utils::durationToSeconds(fullDuration_);
      fullSampleRate_ = fullSampleCount_ / fullDuration;
      fullValueRate_ = fullValueSum_ / fullDuration;
      
      if (fullSampleCount_ > 0) {
        fullValueAverage_ = fullValueSum_ / static_cast<double>(fullSampleCount_);
        
        double squareAvg = fullValueSumOfSquares_ / static_cast<double>(fullSampleCount_);
        double avg = fullValueSum_ / static_cast<double>(fullSampleCount_);
        double sigSquared = squareAvg - avg*avg;
        if(sigSquared > 0.0) {
          fullValueRMS_ = sqrt(sigSquared);
        }
        else {
          fullValueRMS_ = 0.0;
        }
      }
      else {
        fullValueAverage_ = 0.0;
        fullValueRMS_ = 0.0;
      }
      
      // calculate the derived recent values
      const double recentDuration = utils::durationToSeconds(recentDuration_);
      if (recentDuration > 0) {
        recentSampleRate_ = recentSampleCount_ / recentDuration;
        recentValueRate_ = recentValueSum_ / recentDuration;
      }
      else {
        recentSampleRate_ = 0.0;
        recentValueRate_ = 0.0;
      }
      
      if (recentSampleCount_ > 0) {
        recentValueAverage_ = recentValueSum_ / static_cast<double>(recentSampleCount_);
        
        double squareAvg = recentValueSumOfSquares_ /
          static_cast<double>(recentSampleCount_);
        double avg = recentValueSum_ / static_cast<double>(recentSampleCount_);
        double sigSquared = squareAvg - avg*avg;
        if(sigSquared > 0.0) {
          recentValueRMS_ = sqrt(sigSquared);
        }
        else {
          recentValueRMS_ = 0.0;
        }
      }
      else {
        recentValueAverage_ = 0.0;
        recentValueRMS_ = 0.0;
      }
    }
  }

void stor::MonitoredQuantity::disable ( )

Disables the monitor.

Definition at line 293 of file MonitoredQuantity.cc.

References enabled_.

  {
    // It is faster to just set enabled_ to false than to test and set
    // it conditionally.
    enabled_ = false;
  }

void stor::MonitoredQuantity::enable ( )

Enables the monitor (and resets the statistics to provide a fresh start).

Definition at line 285 of file MonitoredQuantity.cc.

References enabled_, and reset().

  {
    if (! enabled_) {
      reset();
      enabled_ = true;
    }
  }

utils::Duration_t stor::MonitoredQuantity::ExpectedCalculationInterval ( ) const [inline]

Definition at line 136 of file MonitoredQuantity.h.

References expectedCalculationInterval_.

    {
      return expectedCalculationInterval_;
    }

void stor::MonitoredQuantity::getStats ( Stats & stats ) const

Write all our collected statistics into the given Stats struct.

Definition at line 343 of file MonitoredQuantity.cc.

Referenced by stor::SMWebPageHelper::addDOMforRunMonitor(), stor::SMWebPageHelper::addDOMforStoredData(), stor::RunMonitorCollection::alarmErrorEvents(), stor::DQMEventMonitorCollection::do_calculateStatistics(), stor::StreamsMonitorCollection::do_calculateStatistics(), stor::FragmentMonitorCollection::do_calculateStatistics(), stor::ThroughputMonitorCollection::do_getStats(), stor::FragmentMonitorCollection::do_updateInfoSpaceItems(), stor::RunMonitorCollection::do_updateInfoSpaceItems(), stor::StreamsMonitorCollection::do_updateInfoSpaceItems(), stor::DQMEventMonitorCollection::getStats(), stor::FragmentMonitorCollection::getStats(), stor::ConsumerMonitorCollection::getTotalStats(), and stor::SMWebPageHelper::listStreamRecordsStats().

  {
    boost::mutex::scoped_lock results(resultsMutex_);
    
    s.fullSampleCount = fullSampleCount_;
    s.fullSampleRate = fullSampleRate_;
    s.fullValueSum = fullValueSum_;
    s.fullValueSumOfSquares = fullValueSumOfSquares_;
    s.fullValueAverage = fullValueAverage_;
    s.fullValueRMS = fullValueRMS_;
    s.fullValueMin = fullValueMin_;
    s.fullValueMax = fullValueMax_;
    s.fullValueRate = fullValueRate_;
    s.fullDuration = fullDuration_;
    
    s.recentSampleCount = recentSampleCount_;
    s.recentSampleRate = recentSampleRate_;
    s.recentValueSum = recentValueSum_;
    s.recentValueSumOfSquares = recentValueSumOfSquares_;
    s.recentValueAverage = recentValueAverage_;
    s.recentValueRMS = recentValueRMS_;
    s.recentValueMin = recentValueMin_;
    s.recentValueMax = recentValueMax_;
    s.recentValueRate = recentValueRate_;
    s.recentDuration = recentDuration_;
    
    s.recentBinnedSampleCounts.resize(binCount_);
    s.recentBinnedValueSums.resize(binCount_);
    s.recentBinnedDurations.resize(binCount_);
    s.recentBinnedSnapshotTimes.resize(binCount_);
    uint32_t sourceBinId = workingBinId_;
    for (uint32_t idx = 0; idx < binCount_; ++idx) {
      if (sourceBinId >= binCount_) {sourceBinId = 0;}
      s.recentBinnedSampleCounts[idx] = binSampleCount_[sourceBinId];
      s.recentBinnedValueSums[idx] = binValueSum_[sourceBinId];
      s.recentBinnedDurations[idx] = binDuration_[sourceBinId];
      s.recentBinnedSnapshotTimes[idx] = binSnapshotTime_[sourceBinId];
      ++sourceBinId;
    }
    
    s.lastSampleValue = lastLatchedSampleValue_;
    s.lastValueRate = lastLatchedValueRate_;
    s.enabled = enabled_;
  }

utils::Duration_t stor::MonitoredQuantity::getTimeWindowForRecentResults ( ) const [inline]

Returns the length of the time window that has been specified for recent results. (This may be different than the actual length of the recent time window which is affected by the interval of calls to the calculateStatistics() method. Use a getDuration(RECENT) call to determine the actual recent time window.)

Definition at line 131 of file MonitoredQuantity.h.

References intervalForRecentStats_.

    {
      return intervalForRecentStats_;
    }

bool stor::MonitoredQuantity::isEnabled ( ) const [inline]

Tests whether the monitor is currently enabled.

Definition at line 115 of file MonitoredQuantity.h.

References enabled_.

{return enabled_;}

MonitoredQuantity& stor::MonitoredQuantity::operator= ( MonitoredQuantity const & ) [private]

void stor::MonitoredQuantity::reset ( void )

Resets the monitor (zeroes out all counters and restarts the time interval).

Definition at line 272 of file MonitoredQuantity.cc.

References accumulationMutex_, resetAccumulators(), resetResults(), and resultsMutex_.

Referenced by stor::RunMonitorCollection::do_reset(), stor::FragmentMonitorCollection::do_reset(), stor::DQMEventMonitorCollection::do_reset(), stor::ThroughputMonitorCollection::do_reset(), stor::StreamsMonitorCollection::do_reset(), stor::ConsumerMonitorCollection::do_reset(), enable(), and stor::ConsumerMonitorCollection::resetCounters().

  {
    {
      boost::mutex::scoped_lock sl(accumulationMutex_);
      resetAccumulators();
    }
    
    {
      boost::mutex::scoped_lock sl(resultsMutex_);
      resetResults();
    }
  }

void stor::MonitoredQuantity::resetAccumulators ( ) [private]

Definition at line 223 of file MonitoredQuantity.cc.

References lastCalculationTime_, workingLastSampleValue_, workingSampleCount_, workingValueMax_, workingValueMin_, workingValueSum_, and workingValueSumOfSquares_.

Referenced by reset(), and setNewTimeWindowForRecentResults().

  {
    lastCalculationTime_ = boost::posix_time::not_a_date_time;
    workingSampleCount_ = 0;
    workingValueSum_ = 0.0;
    workingValueSumOfSquares_ = 0.0;
    workingValueMin_ =  INFINITY;
    workingValueMax_ = -INFINITY;
    workingLastSampleValue_ = 0;
  }

void stor::MonitoredQuantity::resetResults ( ) [private]

Definition at line 234 of file MonitoredQuantity.cc.

References binCount_, binDuration_, binSampleCount_, binSnapshotTime_, binValueMax_, binValueMin_, binValueSum_, binValueSumOfSquares_, fullDuration_, fullSampleCount_, fullSampleRate_, fullValueAverage_, fullValueMax_, fullValueMin_, fullValueRate_, fullValueRMS_, fullValueSum_, fullValueSumOfSquares_, UserOptions_cff::idx, lastLatchedSampleValue_, lastLatchedValueRate_, recentDuration_, recentSampleCount_, recentSampleRate_, recentValueAverage_, recentValueMax_, recentValueMin_, recentValueRate_, recentValueRMS_, recentValueSum_, recentValueSumOfSquares_, seconds(), and workingBinId_.

Referenced by reset(), and setNewTimeWindowForRecentResults().

  {
    workingBinId_ = 0;
    for (unsigned int idx = 0; idx < binCount_; ++idx) {
      binSampleCount_[idx] = 0;
      binValueSum_[idx] = 0.0;
      binValueSumOfSquares_[idx] = 0.0;
      binValueMin_[idx] =  INFINITY;
      binValueMax_[idx] = -INFINITY;
      binDuration_[idx] = boost::posix_time::seconds(0);
      binSnapshotTime_[idx] = boost::posix_time::not_a_date_time;
    }
    
    fullSampleCount_ = 0;
    fullSampleRate_ = 0.0;
    fullValueSum_ = 0.0;
    fullValueSumOfSquares_ = 0.0;
    fullValueAverage_ = 0.0;
    fullValueRMS_ = 0.0;
    fullValueMin_ =  INFINITY;
    fullValueMax_ = -INFINITY;
    fullValueRate_ = 0.0;
    fullDuration_ = boost::posix_time::seconds(0);
    
    recentSampleCount_ = 0;
    recentSampleRate_ = 0.0;
    recentValueSum_ = 0.0;
    recentValueSumOfSquares_ = 0.0;
    recentValueAverage_ = 0.0;
    recentValueRMS_ = 0.0;
    recentValueMin_ =  INFINITY;
    recentValueMax_ = -INFINITY;
    recentValueRate_ = 0.0;
    recentDuration_ = boost::posix_time::seconds(0);
    lastLatchedSampleValue_ = 0.0;
    lastLatchedValueRate_ = 0.0;
  }

void stor::MonitoredQuantity::setNewTimeWindowForRecentResults ( const utils::Duration_t & interval )

Specifies a new time interval to be used when calculating "recent" statistics.

Definition at line 300 of file MonitoredQuantity.cc.

References accumulationMutex_, binCount_, binDuration_, binSampleCount_, binSnapshotTime_, binValueMax_, binValueMin_, binValueSum_, binValueSumOfSquares_, expectedCalculationInterval_, MergeJob_cfg::interval, intervalForRecentStats_, max(), resetAccumulators(), resetResults(), and resultsMutex_.

  {
    // lock the results objects since we're dramatically changing the
    // bins used for the recent results
    {
      boost::mutex::scoped_lock sl(resultsMutex_);
      
      intervalForRecentStats_ = interval;
      
      // determine how many bins we should use in our sliding window
      // by dividing the input time window by the expected calculation
      // interval and rounding to the nearest integer.
      // In case that the calculation interval is larger then the 
      // interval for recent stats, keep the last one.
      binCount_ = std::max(1U,
        static_cast<unsigned int>(
          (intervalForRecentStats_.total_nanoseconds() / expectedCalculationInterval_.total_nanoseconds()) + 0.5
        )      
      );
      
      // create the vectors for the binned quantities
      binSampleCount_.reserve(binCount_);
      binValueSum_.reserve(binCount_);
      binValueSumOfSquares_.reserve(binCount_);
      binValueMin_.reserve(binCount_);
      binValueMax_.reserve(binCount_);
      binDuration_.reserve(binCount_);
      binSnapshotTime_.reserve(binCount_);
      
      resetResults();
    }
    
    {
      boost::mutex::scoped_lock sl(accumulationMutex_);
      resetAccumulators();
    }
    
    // call the reset method to populate the correct initial values
    // for the internal sample data
    //reset();
  }