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/LocalStream.cxx to src/LocalStream.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              :  *  LocalStream.cxx
      14              :  *  ERS
      15              :  *
      16              :  *  Created by Serguei Kolos on 21.01.05.
      17              :  *  Copyright 2005 CERN. All rights reserved.
      18              :  *
      19              :  */
      20              : #include <ers/LocalStream.hpp>
      21              : #include <ers/StreamManager.hpp>
      22              : #include <ers/internal/SingletonCreator.hpp>
      23              : 
      24              : /** This method returns the singleton instance.
      25              :  * It should be used for every operation on the factory.
      26              :  * \return a reference to the singleton instance
      27              :  */
      28              : ers::LocalStream&
      29           41 : ers::LocalStream::instance()
      30              : {
      31              :   /**Singleton instance
      32              :    */
      33           41 :   static ers::LocalStream* instance = ers::SingletonCreator<ers::LocalStream>::create();
      34              : 
      35           41 :   return *instance;
      36              : }
      37              : 
      38              : /** Private constructor - can not be called by user code, use the \c instance() method instead
      39              :  * \see instance()
      40              :  */
      41           15 : ers::LocalStream::LocalStream()
      42           15 :   : m_terminated(false)
      43              : {
      44           15 : }
      45              : 
      46            0 : ers::LocalStream::~LocalStream()
      47              : {
      48            0 :   remove_issue_catcher();
      49            0 : }
      50              : 
      51              : void
      52            0 : ers::LocalStream::remove_issue_catcher()
      53              : {
      54            0 :   std::unique_ptr<std::thread> catcher;
      55            0 :   {
      56            0 :     std::unique_lock lock(m_mutex);
      57            0 :     if (!m_issue_catcher_thread.get()) {
      58            0 :       return;
      59              :     }
      60            0 :     m_terminated = true;
      61            0 :     m_condition.notify_one();
      62            0 :     catcher.swap(m_issue_catcher_thread);
      63            0 :   }
      64              : 
      65            0 :   catcher->join();
      66            0 : }
      67              : 
      68              : void
      69            0 : ers::LocalStream::thread_wrapper()
      70              : {
      71            0 :   std::unique_lock lock(m_mutex);
      72            0 :   m_catcher_thread_id = std::this_thread::get_id();
      73            0 :   while (!m_terminated) {
      74            0 :     m_condition.wait(lock, [this]() { return !m_issues.empty() || m_terminated; });
      75              : 
      76            0 :     while (!m_terminated && !m_issues.empty()) {
      77            0 :       ers::Issue* issue = m_issues.front();
      78            0 :       m_issues.pop();
      79              : 
      80            0 :       lock.unlock();
      81            0 :       m_issue_catcher(*issue);
      82            0 :       delete issue;
      83            0 :       lock.lock();
      84              :     }
      85              :   }
      86            0 :   m_catcher_thread_id = {};
      87            0 :   m_terminated = false;
      88            0 : }
      89              : 
      90              : ers::IssueCatcherHandler*
      91            0 : ers::LocalStream::set_issue_catcher(const std::function<void(const ers::Issue&)>& catcher)
      92              : {
      93            0 :   std::unique_lock lock(m_mutex);
      94            0 :   if (m_issue_catcher_thread.get()) {
      95            0 :     throw ers::IssueCatcherAlreadySet(ERS_HERE);
      96              :   }
      97            0 :   m_issue_catcher = catcher;
      98            0 :   m_issue_catcher_thread.reset(new std::thread(std::bind(&ers::LocalStream::thread_wrapper, this)));
      99              : 
     100            0 :   return new ers::IssueCatcherHandler;
     101            0 : }
     102              : 
     103              : void
     104           41 : ers::LocalStream::report_issue(ers::severity type, const ers::Issue& issue)
     105              : {
     106           41 :   if (m_issue_catcher_thread.get() && m_catcher_thread_id != std::this_thread::get_id()) {
     107            0 :     ers::Issue* clone = issue.clone();
     108            0 :     clone->set_severity(type);
     109            0 :     std::unique_lock lock(m_mutex);
     110            0 :     m_issues.push(clone);
     111            0 :     m_condition.notify_one();
     112            0 :   } else {
     113           41 :     StreamManager::instance().report_issue(type, issue);
     114              :   }
     115           41 : }
     116              : 
     117              : void
     118           27 : ers::LocalStream::error(const ers::Issue& issue)
     119              : {
     120           27 :   report_issue(ers::Error, issue);
     121           27 : }
     122              : 
     123              : void
     124            0 : ers::LocalStream::fatal(const ers::Issue& issue)
     125              : {
     126            0 :   report_issue(ers::Fatal, issue);
     127            0 : }
     128              : 
     129              : void
     130           14 : ers::LocalStream::warning(const ers::Issue& issue)
     131              : {
     132           14 :   report_issue(ers::Warning, issue);
     133           14 : }