LocalStream.cpp

Go to the documentation of this file.

/*
 * DUNE DAQ modification notice:
 * This file has been modified from the original ATLAS ers source for the DUNE DAQ project.
 * Fork baseline commit: 8267df82a4f6fe6bf02c4014923eba19eddc4614 (2020-04-14).
 * Renamed since fork: yes (from src/LocalStream.cxx to src/LocalStream.cpp).
 *
 * Original copyright:
 * Copyright (C) 2001-2020 CERN for the benefit of the ATLAS collaboration.
 * Licensed under the Apache License, Version 2.0.
 */
 
/*
 *  LocalStream.cxx
 *  ERS
 *
 *  Created by Serguei Kolos on 21.01.05.
 *  Copyright 2005 CERN. All rights reserved.
 *
 */
#include <ers/LocalStream.hpp>
#include <ers/StreamManager.hpp>
#include <ers/internal/SingletonCreator.hpp>

ers::LocalStream &
ers::LocalStream::instance()
{
    static ers::LocalStream * instance = ers::SingletonCreator<ers::LocalStream>::create();
 
    return *instance;
}

ers::LocalStream::LocalStream( )
  : m_terminated( false )
{ }
 
ers::LocalStream::~LocalStream( )
{
    remove_issue_catcher();
}
 
void
ers::LocalStream::remove_issue_catcher( )
{
    std::unique_ptr<std::thread> catcher;
    {
    std::unique_lock lock( m_mutex );
    if ( !m_issue_catcher_thread.get() )
    {
        return ;
    }
    m_terminated = true;
    m_condition.notify_one();
    catcher.swap(m_issue_catcher_thread);
    }
    
    catcher -> join();
}
 
void
ers::LocalStream::thread_wrapper()
{
    std::unique_lock lock( m_mutex );
    m_catcher_thread_id = std::this_thread::get_id();
    while( !m_terminated )
    {
    m_condition.wait( lock, [this](){return !m_issues.empty() || m_terminated;} );
        
        while( !m_terminated && !m_issues.empty() )
        {
            ers::Issue * issue = m_issues.front();
            m_issues.pop();
            
            lock.unlock();            
            m_issue_catcher( *issue );
        delete issue;
            lock.lock();            
        }
    }
    m_catcher_thread_id = {};
    m_terminated = false;
}
 
ers::IssueCatcherHandler *
ers::LocalStream::set_issue_catcher( const std::function<void ( const ers::Issue & )> & catcher )
{
    std::unique_lock lock( m_mutex );
    if ( m_issue_catcher_thread.get() )
    {
        throw ers::IssueCatcherAlreadySet( ERS_HERE );
    }
    m_issue_catcher = catcher;
    m_issue_catcher_thread.reset( new std::thread( std::bind( &ers::LocalStream::thread_wrapper, this ) ) );
    
    return new ers::IssueCatcherHandler;
}
 
void 
ers::LocalStream::report_issue( ers::severity type, const ers::Issue & issue )
{
    if ( m_issue_catcher_thread.get() && m_catcher_thread_id != std::this_thread::get_id() )
    {
    ers::Issue * clone = issue.clone();
    clone->set_severity( type );
    std::unique_lock lock( m_mutex );
    m_issues.push( clone );
    m_condition.notify_one();
    }
    else
    {
    StreamManager::instance().report_issue( type, issue );
    }
}
 
void 
ers::LocalStream::error( const ers::Issue & issue )
{
    report_issue( ers::Error, issue );
}
 
void 
ers::LocalStream::fatal( const ers::Issue & issue )
{
    report_issue( ers::Fatal, issue );
}
 
void 
ers::LocalStream::warning( const ers::Issue & issue )
{
    report_issue( ers::Warning, issue );
}