BaseSink.h

Go to the documentation of this file.

/***********************************************************************************\
* (c) Copyright 2022 CERN for the benefit of the LHCb and ATLAS collaborations      *
*                                                                                   *
* This software is distributed under the terms of the Apache version 2 licence,     *
* copied verbatim in the file "LICENSE".                                            *
*                                                                                   *
* In applying this licence, CERN does not waive the privileges and immunities       *
* granted to it by virtue of its status as an Intergovernmental Organization        *
* or submit itself to any jurisdiction.                                             *
\***********************************************************************************/
 
#include <Gaudi/MonitoringHub.h>
#include <Gaudi/Property.h>
#include <GaudiKernel/Service.h>
 
#include <chrono>
#include <future>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
 
namespace Gaudi::Monitoring {
 
  class BaseSink : public Service, public Hub::Sink {
 
  public:
    using Service::Service;
 
    StatusCode initialize() override {
      // registers itself to the Monitoring Hub
      return Service::initialize().andThen( [&] { serviceLocator()->monitoringHub().addSink( this ); } );
    }
 
    void registerEntity( Hub::Entity ent ) override {
      if ( wanted( ent.type, m_typesToSave ) && wanted( ent.name, m_namesToSave ) &&
           wanted( ent.component, m_componentsToSave ) ) {
        m_monitoringEntities.emplace( ent.id(), std::move( ent ) );
      }
    }
 
    void removeEntity( Hub::Entity const& ent ) override {
      auto it = m_monitoringEntities.find( ent.id() );
      if ( it != m_monitoringEntities.end() ) { m_monitoringEntities.erase( it ); }
    }
 
    virtual void flush( bool isStop ) = 0;
 
    StatusCode start() override {
      return Service::start().andThen( [&] {
        // promise needs to be recreated in case of a restart
        m_flushThreadStop = std::promise<void>{};
        // enable periodic output file flush if requested
        if ( m_autoFlushPeriod.value() > std::numeric_limits<double>::epsilon() ) {
          m_flushThread = std::thread{ [this, flushStop = m_flushThreadStop.get_future()]() {
            using namespace std::chrono_literals;
            while ( flushStop.wait_for( m_autoFlushPeriod.value() * 1s ) == std::future_status::timeout ) {
              flush( false );
            }
          } };
        }
      } );
    }
 
    StatusCode stop() override {
      m_flushThreadStop.set_value();                        // tell the flush thread we are stopping
      if ( m_flushThread.joinable() ) m_flushThread.join(); // and wait that it exits
      flush( true );
      return Service::stop();
    }
 
  protected:
    template <typename Callable>
    void applyToAllEntities( Callable func ) const {
      std::for_each( begin( m_monitoringEntities ), end( m_monitoringEntities ),
                     [func]( auto& p ) { func( p.second ); } );
    }
 
    std::map<std::string, std::map<std::string, nlohmann::json>> sortedEntitiesAsJSON() const {
      std::map<std::string, std::map<std::string, nlohmann::json>> sortedEntities;
      applyToAllEntities( [&sortedEntities]( auto& ent ) { sortedEntities[ent.component][ent.name] = ent; } );
      return sortedEntities;
    }
 
    bool wanted( std::string name, std::vector<std::string> searchNames ) {
      if ( searchNames.empty() ) { return true; }
      for ( const auto& searchName : searchNames ) {
        const std::regex regex( searchName );
        if ( std::regex_match( name, regex ) ) { return true; }
      }
      return false;
    }
 
    std::map<void*, Gaudi::Monitoring::Hub::Entity> m_monitoringEntities;
    Gaudi::Property<std::vector<std::string>>       m_namesToSave{
        this, "NamesToSave", {}, "List of regexps used to match names of entities to save" };
    Gaudi::Property<std::vector<std::string>> m_componentsToSave{
        this, "ComponentsToSave", {}, "List of regexps used to match component names of entities to save" };
    Gaudi::Property<std::vector<std::string>> m_typesToSave{
        this, "TypesToSave", {}, "List of regexps used to match type names of entities to save" };
 
    std::thread            m_flushThread;
    std::promise<void>     m_flushThreadStop;
    Gaudi::Property<float> m_autoFlushPeriod{
        this, "AutoFlushPeriod", 0.,
        "if different from 0, indicates every how many seconds to force a write of the FSR data to OutputFile (this "
        "parameter makes sense only if used in conjunction with OutputFile)" };
  };
 
} // namespace Gaudi::Monitoring