#include <ProcessorMetricCollector.hpp>

Public Types
using	signal_t = T

Public Member Functions
void	attach_processor (AbstractProcessor< signal_t > &processor, size_t pipeline_id)

void	configure (const std::vector< std::pair< std::string, nlohmann::json > > configs, const std::vector< std::pair< dunedaq::trgdataformats::channel_t, int16_t > > channel_plane_numbers, uint8_t num_pipelines)

std::vector< AbstractProcessor< signal_t > * >	_get_attached_processors ()

std::vector< ProcessorMetricInformation >	_get_processor_metric_table ()

std::vector< std::vector< std::vector< int16_t > > >	_get_processor_casted_data_table ()

void	signal_collect ()

void	run ()

std::vector< std::vector< signal_t > >	_get_retrieved_processor_metrics () const

std::unordered_map< dunedaq::trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > >	get_metrics ()

void	stop ()

	ProcessorMetricCollector ()

void	lock_metric_modify ()

void	unlock_metric_modify ()

Private Member Functions
void	cast_metrics_from_raw_type ()

void	collect_metrics_from_attached_processors ()

void	convert_into_channel_metric_value ()

Private Attributes
std::vector< AbstractProcessor< signal_t > * >	m_attached_processors

std::vector< ProcessorMetricInformation >	m_processor_metric_table

std::vector< std::vector< std::vector< int16_t > > >	m_processor_casted_data_table

std::vector< std::vector< signal_t > >	m_processor_metric_collection_table

std::vector< dunedaq::trgdataformats::channel_t >	m_channel_numbers

std::unordered_map< dunedaq::trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > >	m_metrics

std::mutex	m_mutex

std::atomic< bool >	m_signal_collect {false}

std::thread	m_collector_thread

std::atomic< bool >	m_stop_flag {false}

size_t	m_attach_counter = 0

Detailed Description

template<typename T>
class tpglibs::ProcessorMetricCollector< T >

Definition at line 36 of file ProcessorMetricCollector.hpp.

Member Typedef Documentation

◆ signal_t

template<typename T >

using tpglibs::ProcessorMetricCollector< T >::signal_t = T

Definition at line 39 of file ProcessorMetricCollector.hpp.

Constructor & Destructor Documentation

◆ ProcessorMetricCollector()

template<typename T >

tpglibs::ProcessorMetricCollector< T >::ProcessorMetricCollector ( )

inline

Definition at line 84 of file ProcessorMetricCollector.hpp.

    : m_signal_collect(false)
    , m_stop_flag(false)
    , m_attach_counter(0) {
  }

Member Function Documentation

◆ _get_attached_processors()

template<typename signal_t >

std::vector< AbstractProcessor< signal_t > * > tpglibs::ProcessorMetricCollector< signal_t >::_get_attached_processors ( )

Definition at line 22 of file ProcessorMetricCollector.cpp.

                                                                                                     {
  return m_attached_processors;
}

◆ _get_processor_casted_data_table()

template<typename signal_t >

std::vector< std::vector< std::vector< int16_t > > > tpglibs::ProcessorMetricCollector< signal_t >::_get_processor_casted_data_table ( )

Definition at line 32 of file ProcessorMetricCollector.cpp.

                                                                                                              {
  return m_processor_casted_data_table;
}

◆ _get_processor_metric_table()

template<typename signal_t >

std::vector< ProcessorMetricInformation > tpglibs::ProcessorMetricCollector< signal_t >::_get_processor_metric_table ( )

Definition at line 27 of file ProcessorMetricCollector.cpp.

                                                                                                      {
  return m_processor_metric_table;
}

◆ _get_retrieved_processor_metrics()

template<typename signal_t >

std::vector< std::vector< signal_t > > tpglibs::ProcessorMetricCollector< signal_t >::_get_retrieved_processor_metrics ( ) const

Definition at line 150 of file ProcessorMetricCollector.cpp.

                                                                           {
  // TODO: return collected metrics
  return m_processor_metric_collection_table;
}

◆ attach_processor()

template<typename T >

void tpglibs::ProcessorMetricCollector< T >::attach_processor	(	AbstractProcessor< signal_t > &	processor,
		size_t	pipeline_id )

inline

Definition at line 41 of file ProcessorMetricCollector.hpp.

                                                                                    {
    // Attach a processor to be observed (collected) by this
    m_attached_processors[m_attach_counter] = &processor;
 
    // Look up and fill in information about this processor: its pipeline belonging and what is collected
    auto metrics = processor.get_metric_items();
    m_processor_metric_table[m_attach_counter].m_names_of_metrics = metrics;
    m_processor_metric_table[m_attach_counter].m_pipeline_id = pipeline_id;
 
    m_processor_casted_data_table.emplace_back(
        metrics.size(),
        std::vector<int16_t>(16)
    );
 
    // Then instantiate the space for storing collected metrics
    // Preallocated to preconfigured number of metrics
    m_processor_metric_collection_table.push_back(std::vector<signal_t>(metrics.size()));
 
    for (size_t i = pipeline_id*16; i < (pipeline_id+1)*16; i++) {
      for (size_t j = 0; j < metrics.size(); j++) {
        // Record metric name and initial value 0 for this channel
        m_metrics[m_channel_numbers[i]].emplace_back(metrics[j], int16_t{0});
      }
    }
 
    m_attach_counter++;
  }

◆ cast_metrics_from_raw_type()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::cast_metrics_from_raw_type ( )

private

Definition at line 90 of file ProcessorMetricCollector.cpp.

                                                                    {
  for (size_t i = 0; i < m_processor_metric_collection_table.size(); i++) {
    auto & raw = m_processor_metric_collection_table[i];
    for (size_t j = 0; j < raw.size(); j++) {
      int16_t out[16];
      _mm256_storeu_si256(reinterpret_cast<__m256i*>(&out), raw[j]);
      for (size_t k = 0; k < 16; k++) {
        m_processor_casted_data_table[i][j][k] = out[k];
      }
    }
  }
}

◆ collect_metrics_from_attached_processors()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::collect_metrics_from_attached_processors ( )

private

Definition at line 77 of file ProcessorMetricCollector.cpp.

                                                                                  {
  size_t proc_id = 0;
  for (auto processor_ptr : m_attached_processors) {
    if (!processor_ptr) continue;
    ProcessorMetricArray items = processor_ptr->read_from_metric_store_buffer();
    for (size_t i = 0; i < items.m_size; i++) {
      m_processor_metric_collection_table[proc_id][i] = items.m_data[i];
    }
    proc_id++;
  }
}

◆ configure()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::configure	(	const std::vector< std::pair< std::string, nlohmann::json > >	configs,
		const std::vector< std::pair< dunedaq::trgdataformats::channel_t, int16_t > >	channel_plane_numbers,
		uint8_t	num_pipelines )

Definition at line 37 of file ProcessorMetricCollector.cpp.

{
  // Stop any existing thread first - use proper synchronization
  if (m_collector_thread.joinable()) {
    m_stop_flag.store(true, std::memory_order_relaxed);
    // Give the thread a chance to see the stop flag
    std::this_thread::yield();
    m_collector_thread.join();
  }
 
  // Reset all state for fresh start
  m_signal_collect.store(false, std::memory_order_relaxed);
  m_stop_flag.store(false, std::memory_order_relaxed);
  m_attach_counter = 0;
 
  // Clear all data structures
  m_metrics.clear();
 
  // create a fixed size array for reference to all the processors
  // configs are pairs of (processor name, specific configs)
  size_t n_processors = configs.size();
  m_attached_processors = std::vector<AbstractProcessor<signal_t>*>(n_processors * num_pipelines);
 
  m_processor_metric_table = std::vector<ProcessorMetricInformation>(n_processors * num_pipelines, ProcessorMetricInformation{0, {}});
 
  m_processor_metric_collection_table = {};
  
  m_processor_casted_data_table = {};
 
  m_channel_numbers = std::vector<dunedaq::trgdataformats::channel_t>(channel_plane_numbers.size(), 0);
 
  for (size_t i = 0; i < channel_plane_numbers.size(); i++) {
    m_channel_numbers[i] = channel_plane_numbers[i].first;
    m_metrics[m_channel_numbers[i]] = std::vector<std::pair<std::string, int16_t>>();
  }
}

◆ convert_into_channel_metric_value()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::convert_into_channel_metric_value ( )

private

Definition at line 172 of file ProcessorMetricCollector.cpp.

                                                                           {
  std::scoped_lock lock(m_mutex);
  std::vector<int16_t> mcount(m_channel_numbers.size(), 0);
 
  for (size_t pid = 0; pid < m_processor_casted_data_table.size(); pid++) {
    for (size_t mid = 0; mid < m_processor_casted_data_table[pid].size(); mid++) {
      auto responsible_channels_start = m_processor_metric_table[pid].m_pipeline_id * 16;
      size_t local_cid_in_table=0;
      for (size_t cid = responsible_channels_start; cid < responsible_channels_start + 16; cid++) {
        auto mnames = m_processor_metric_table[pid].m_names_of_metrics;
        m_metrics[m_channel_numbers[cid]][mcount[cid]].second = m_processor_casted_data_table[pid][mid][local_cid_in_table];
        m_metrics[m_channel_numbers[cid]][mcount[cid]].first = m_processor_metric_table[pid].m_names_of_metrics[mid];
        local_cid_in_table++;
        mcount[cid]++;
      }
    }
  }
}

◆ get_metrics()

template<typename signal_t >

std::unordered_map< dunedaq::trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > > tpglibs::ProcessorMetricCollector< signal_t >::get_metrics ( )

Definition at line 192 of file ProcessorMetricCollector.cpp.

                                                                                                                                         {
  return m_metrics;
}

◆ lock_metric_modify()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::lock_metric_modify ( )

Definition at line 197 of file ProcessorMetricCollector.cpp.

                                                            {
  m_mutex.lock();
}

◆ run()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::run ( )

Definition at line 113 of file ProcessorMetricCollector.cpp.

                                             {
  // TODO: main loop for metric collection thread
  // Check if thread is already running
  m_stop_flag.store(false, std::memory_order_release);
  if (m_collector_thread.joinable()) {
    return; // Thread already running
  }
  
  m_collector_thread = std::thread([this]() {
    while (!this->m_stop_flag.load(std::memory_order_acquire)) {
      // wait on collect signal
      m_signal_collect.wait(false, std::memory_order_relaxed);
 
      // when woken up, check if stop flag is set
 
      if (this->m_stop_flag.load(std::memory_order_acquire)) {
        break;
      }
 
      // If this is signaled to collect metrics
      this->collect_metrics_from_attached_processors();
 
      this->m_signal_collect.store(false, std::memory_order_relaxed);
      // After collection, we reset the collect flag to false.
      // Note that when signal_collect() is called at a far higher rate then possible, ultimately collection
      // happens at the highest possible rate, not necessarily the set rate
 
      // perform any post processings on the collected metrics as needed
      this->cast_metrics_from_raw_type();
      this->convert_into_channel_metric_value();
      
    }
  });
}

◆ signal_collect()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::signal_collect ( )

Definition at line 104 of file ProcessorMetricCollector.cpp.

                                                        {
  // TODO: signal that a collection cycle should occur
  m_signal_collect.store(true, std::memory_order_relaxed);
  m_signal_collect.notify_one();
  // this->collect_metrics_from_attached_processors();
  // this->cast_metrics_from_raw_type();
}

◆ stop()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::stop ( )

Definition at line 156 of file ProcessorMetricCollector.cpp.

                                              {
  // TODO: stop the collection thread
  m_stop_flag.store(true, std::memory_order_release);
  m_signal_collect.store(true, std::memory_order_release);
  // make sure stop flag is set
  while (!this->m_stop_flag.load(std::memory_order_acquire)) {
    std::this_thread::yield();
  }
  // notify the thread to wake up
  m_signal_collect.notify_one();
  if (m_collector_thread.joinable()) {
    m_collector_thread.join();
  }
}

◆ unlock_metric_modify()

template<typename signal_t >

void tpglibs::ProcessorMetricCollector< signal_t >::unlock_metric_modify ( )

Definition at line 202 of file ProcessorMetricCollector.cpp.

                                                              {
  m_mutex.unlock();
}

Member Data Documentation

◆ m_attach_counter

template<typename T >

size_t tpglibs::ProcessorMetricCollector< T >::m_attach_counter = 0

private

Definition at line 111 of file ProcessorMetricCollector.hpp.

◆ m_attached_processors

template<typename T >

std::vector<AbstractProcessor<signal_t>*> tpglibs::ProcessorMetricCollector< T >::m_attached_processors

private

Definition at line 98 of file ProcessorMetricCollector.hpp.

◆ m_channel_numbers

template<typename T >

std::vector<dunedaq::trgdataformats::channel_t> tpglibs::ProcessorMetricCollector< T >::m_channel_numbers

private

Definition at line 103 of file ProcessorMetricCollector.hpp.

◆ m_collector_thread

template<typename T >

std::thread tpglibs::ProcessorMetricCollector< T >::m_collector_thread

private

Definition at line 109 of file ProcessorMetricCollector.hpp.

◆ m_metrics

template<typename T >

std::unordered_map<dunedaq::trgdataformats::channel_t, std::vector<std::pair<std::string, int16_t> > > tpglibs::ProcessorMetricCollector< T >::m_metrics

private

Definition at line 105 of file ProcessorMetricCollector.hpp.

◆ m_mutex

template<typename T >

std::mutex tpglibs::ProcessorMetricCollector< T >::m_mutex

private

Definition at line 106 of file ProcessorMetricCollector.hpp.

◆ m_processor_casted_data_table

template<typename T >

std::vector<std::vector<std::vector<int16_t> > > tpglibs::ProcessorMetricCollector< T >::m_processor_casted_data_table

private

Definition at line 100 of file ProcessorMetricCollector.hpp.

◆ m_processor_metric_collection_table

template<typename T >

std::vector<std::vector<signal_t> > tpglibs::ProcessorMetricCollector< T >::m_processor_metric_collection_table

private

Definition at line 101 of file ProcessorMetricCollector.hpp.

◆ m_processor_metric_table

template<typename T >

std::vector<ProcessorMetricInformation> tpglibs::ProcessorMetricCollector< T >::m_processor_metric_table

private

Definition at line 99 of file ProcessorMetricCollector.hpp.

◆ m_signal_collect

template<typename T >

std::atomic<bool> tpglibs::ProcessorMetricCollector< T >::m_signal_collect {false}

private

Definition at line 108 of file ProcessorMetricCollector.hpp.

108{false};

◆ m_stop_flag

template<typename T >

std::atomic<bool> tpglibs::ProcessorMetricCollector< T >::m_stop_flag {false}

private

Definition at line 110 of file ProcessorMetricCollector.hpp.

110{false};

The documentation for this class was generated from the following files:

/github/workspace/dunedaq/sourcecode/tpglibs/include/tpglibs/ProcessorMetricCollector.hpp
/github/workspace/dunedaq/sourcecode/tpglibs/src/ProcessorMetricCollector.cpp

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ signal_t

Constructor & Destructor Documentation

◆ ProcessorMetricCollector()

Member Function Documentation

◆ _get_attached_processors()

◆ _get_processor_casted_data_table()

◆ _get_processor_metric_table()

◆ _get_retrieved_processor_metrics()

◆ attach_processor()

◆ cast_metrics_from_raw_type()

◆ collect_metrics_from_attached_processors()

◆ configure()

◆ convert_into_channel_metric_value()

◆ get_metrics()

◆ lock_metric_modify()

◆ run()

◆ signal_collect()

◆ stop()

◆ unlock_metric_modify()

Member Data Documentation

◆ m_attach_counter

◆ m_attached_processors

◆ m_channel_numbers

◆ m_collector_thread

◆ m_metrics

◆ m_mutex

◆ m_processor_casted_data_table

◆ m_processor_metric_collection_table

◆ m_processor_metric_table

◆ m_signal_collect

◆ m_stop_flag