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/StreamFactory.hpp>
      23              : #include <ers/internal/FilterStream.hpp>
      24              : #include <ers/internal/Util.hpp>
      25              : 
      26              : #include <ers/internal/ThrottleStream.hpp>
      27              : 
      28           66 : ERS_REGISTER_OUTPUT_STREAM(ers::ThrottleStream, "throttle", format)
      29              : 
      30           33 : ers::ThrottleStream::IssueRecord::IssueRecord()
      31              : {
      32           33 :   reset();
      33           33 : }
      34              : 
      35              : void
      36           66 : ers::ThrottleStream::IssueRecord::reset()
      37              : {
      38           66 :   m_lastOccurance = 0;
      39           66 :   m_lastReport = 0;
      40           66 :   m_initialCounter = 0;
      41           66 :   m_threshold = 10;
      42           66 :   m_suppressedCounter = 0;
      43           66 : }
      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           60 : ers::ThrottleStream::throttle(IssueRecord& rec, const ers::Issue& issue)
      65              : {
      66           60 :   std::time_t issueTime = issue.time_t();
      67           60 :   bool reported = false;
      68           60 :   if (issueTime - rec.m_lastOccurance > m_timeLimit) {
      69           33 :     if (rec.m_suppressedCounter > 0) {
      70            0 :       reportSuppression(rec, issue);
      71            0 :       reported = true;
      72              :     }
      73           33 :     rec.reset();
      74              :   }
      75              : 
      76           60 :   if (rec.m_initialCounter < m_initialThreshold) {
      77           60 :     rec.m_initialCounter++;
      78           60 :     rec.m_lastReport = issueTime;
      79           60 :     if (!reported) {
      80           60 :       chained().write(issue);
      81              :     }
      82            0 :   } else if (rec.m_suppressedCounter >= rec.m_threshold) {
      83            0 :     rec.m_threshold = rec.m_threshold * 10;
      84            0 :     reportSuppression(rec, issue);
      85            0 :   } else if (issueTime - rec.m_lastReport > m_timeLimit) {
      86            0 :     reportSuppression(rec, issue);
      87              :   } else {
      88            0 :     rec.m_suppressedCounter++;
      89              :   }
      90              : 
      91           60 :   rec.m_lastOccurance = issueTime;
      92           60 :   rec.m_lastOccuranceFormatted = issue.time<std::chrono::microseconds>();
      93           60 : }
      94              : 
      95           23 : ers::ThrottleStream::ThrottleStream(const std::string& criteria)
      96              : {
      97           23 :   m_initialThreshold = 30;
      98           23 :   m_timeLimit = 30;
      99              : 
     100           23 :   std::vector<std::string> params;
     101           23 :   ers::tokenize(criteria, ",", params);
     102              : 
     103           23 :   if (params.size() > 0) {
     104           23 :     std::istringstream in(params[0]);
     105           23 :     in >> m_initialThreshold;
     106           23 :   }
     107              : 
     108           23 :   if (params.size() > 1) {
     109            0 :     std::istringstream in(params[1]);
     110            0 :     in >> m_timeLimit;
     111            0 :   }
     112           23 : }
     113              : 
     114              : /** Write method
     115              :  * basically calls \c throttle to check if the issue is accepted.
     116              :  * If this is the case, the \c write method on the chained stream is called with
     117              :  * \c issue.
     118              :  * \param issue issue to be sent.
     119              :  */
     120              : void
     121           60 : ers::ThrottleStream::write(const ers::Issue& issue)
     122              : {
     123           60 :   const ers::Context& context = issue.context();
     124           60 :   std::string issueId = context.file_name() + boost::lexical_cast<std::string>(context.line_number());
     125              : 
     126           60 :   std::scoped_lock ml(m_mutex);
     127           60 :   throttle(m_issueMap[issueId], issue);
     128           60 : }