Line data    Source code

       1              : /*
       2              :  * DUNE DAQ modification notice:
       3              :  * This file has been modified from the original ATLAS ers source for the DUNE DAQ project.
       4              :  * Fork baseline commit: 8267df82a4f6fe6bf02c4014923eba19eddc4614 (2020-04-14).
       5              :  * Renamed since fork: yes (from src/streams/ThrottleStream.cxx to plugins/ThrottleStream.cpp).
       6              :  *
       7              :  * Original copyright:
       8              :  * Copyright (C) 2001-2020 CERN for the benefit of the ATLAS collaboration.
       9              :  * Licensed under the Apache License, Version 2.0.
      10              :  */
      11              : 
      12              : /*
      13              :  *  ThrottleStream.cxx
      14              :  *  ers
      15              :  *
      16              :  *  Created by Gordon Crone on 02.08.05.
      17              :  *  Copyright 2004 CERN. All rights reserved.
      18              :  *
      19              :  */
      20              : #include <boost/lexical_cast.hpp>
      21              : 
      22              : #include <ers/internal/FilterStream.hpp>
      23              : #include <ers/internal/Util.hpp>
      24              : #include <ers/StreamFactory.hpp>
      25              : 
      26              : #include <ers/internal/ThrottleStream.hpp>
      27              : 
      28           64 : ERS_REGISTER_OUTPUT_STREAM( ers::ThrottleStream, "throttle", format )
      29              : 
      30           32 : ers::ThrottleStream::IssueRecord::IssueRecord()
      31              : {
      32           32 :     reset();
      33           32 : }
      34              : 
      35              : void 
      36           64 : ers::ThrottleStream::IssueRecord::reset()
      37              : {
      38           64 :     m_lastOccurance=0;
      39           64 :     m_lastReport=0;
      40           64 :     m_initialCounter=0;
      41           64 :     m_threshold=10;
      42           64 :     m_suppressedCounter=0;
      43           64 : }
      44              : 
      45              : void 
      46            0 : ers::ThrottleStream::reportSuppression(IssueRecord& record, const ers::Issue& issue)
      47              : {
      48            0 :     std::ostringstream msgStream;
      49            0 :     msgStream << " -- " << record.m_suppressedCounter << " similar messages suppressed, last occurrence was at "
      50            0 :                 << record.m_lastOccuranceFormatted;
      51              :     
      52            0 :     ers::Issue* suppressedNotice = issue.clone();
      53            0 :     suppressedNotice->wrap_message( "",  msgStream.str());
      54              : 
      55            0 :     suppressedNotice->set_severity(issue.severity());
      56            0 :     chained().write(*suppressedNotice);
      57            0 :     delete suppressedNotice;
      58              : 
      59            0 :     record.m_lastReport = issue.time_t();
      60            0 :     record.m_suppressedCounter = 0;
      61            0 : }
      62              : 
      63              : void 
      64           59 : ers::ThrottleStream::throttle(IssueRecord& rec, const ers::Issue& issue)
      65              : {
      66           59 :     std::time_t issueTime=issue.time_t();
      67           59 :     bool reported=false;
      68           59 :     if (issueTime - rec.m_lastOccurance > m_timeLimit) {
      69           32 :         if (rec.m_suppressedCounter>0) {
      70            0 :            reportSuppression(rec, issue);
      71            0 :            reported=true;
      72              :         }
      73           32 :         rec.reset();
      74              :     }
      75              : 
      76           59 :     if (rec.m_initialCounter<m_initialThreshold) {
      77           59 :         rec.m_initialCounter++;
      78           59 :         rec.m_lastReport=issueTime;
      79           59 :         if (!reported) {
      80           59 :             chained().write(issue);
      81              :         }
      82              :     }
      83            0 :     else if (rec.m_suppressedCounter>=rec.m_threshold) {
      84            0 :         rec.m_threshold=rec.m_threshold*10;
      85            0 :         reportSuppression(rec, issue);
      86              :     }
      87            0 :     else if (issueTime - rec.m_lastReport > m_timeLimit) {
      88            0 :         reportSuppression(rec, issue);
      89              :     }
      90              :     else {
      91            0 :         rec.m_suppressedCounter++;
      92              :     }
      93              : 
      94           59 :     rec.m_lastOccurance=issueTime;
      95           59 :     rec.m_lastOccuranceFormatted=issue.time<std::chrono::microseconds>();
      96           59 : }
      97              : 
      98           22 : ers::ThrottleStream::ThrottleStream( const std::string & criteria )
      99              : {
     100           22 :     m_initialThreshold = 30;
     101           22 :     m_timeLimit = 30;
     102              :     
     103           22 :     std::vector<std::string> params;
     104           22 :     ers::tokenize( criteria, ",", params );
     105              :     
     106           22 :     if ( params.size() > 0 )
     107              :     {
     108           22 :         std::istringstream in( params[0] );
     109           22 :         in >> m_initialThreshold;
     110           22 :     }
     111              :     
     112           22 :     if ( params.size() > 1 )
     113              :     {
     114            0 :         std::istringstream in( params[1] );
     115            0 :         in >> m_timeLimit;
     116            0 :     }
     117           22 : }
     118              : 
     119              : /** Write method 
     120              :   * basically calls \c throttle to check if the issue is accepted. 
     121              :   * If this is the case, the \c write method on the chained stream is called with 
     122              :   * \c issue. 
     123              :   * \param issue issue to be sent.
     124              :   */
     125              : void 
     126           59 : ers::ThrottleStream::write( const ers::Issue & issue )
     127              : {
     128           59 :     const ers::Context& context = issue.context();
     129           59 :     std::string issueId = context.file_name() + boost::lexical_cast<std::string>(context.line_number());
     130              : 
     131           59 :     std::scoped_lock ml(m_mutex);
     132           59 :     throttle( m_issueMap[issueId], issue );
     133           59 : }