dunedaq::fdreadoutlibs::TPGInternalStateHarvester Class Reference

#include <TPGInternalStateHarvester.hpp>

Public Types
using	ProcRef = std::pair<std::shared_ptr<tpglibs::AbstractProcessor<__m256i>>, int >

Public Member Functions
	~TPGInternalStateHarvester ()
void	set_processor_references (std::vector< ProcRef > refs)
const std::vector< ProcRef > &	get_processor_references () const
void	update_channel_plane_numbers (const std::vector< std::pair< trgdataformats::channel_t, int16_t > > &channel_plane_numbers, uint8_t num_channels_per_pipeline, uint8_t num_pipelines)
	Cuts a full list of (channel, plane) into per-pipeline lists of 16 lanes each.
std::unordered_map< trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > >	harvest_once ()
	Harvest once, outputs channel -> [(metric_name, value)...].
void	start_collection_thread ()
	Start the background collection thread Must be called before using async collection features.
void	stop_collection_thread ()
	Stop the background collection thread Blocks until thread is fully stopped.
void	trigger_harvest ()
	Signal the collection thread to perform one harvest cycle Non-blocking - returns immediately.
std::unordered_map< trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > >	get_latest_results () const
	Get the latest collected results (thread-safe, non-blocking) Returns a copy of the most recent harvest results.
bool	is_collection_thread_running () const
	Check if collection thread is running.

Private Types
using	ResultContainer

Private Member Functions
void	collection_thread_worker_ ()
void	rebuild_prealloc_caches_ ()

Private Attributes
std::vector< ProcRef >	m_processor_references
std::vector< std::vector< std::pair< trgdataformats::channel_t, int16_t > > >	m_channel_plane_numbers_per_pipeline
uint8_t	m_num_channels_per_pipeline
uint8_t	m_num_pipelines
std::vector< std::vector< std::string > >	m_metric_items_per_proc
std::vector< size_t >	m_expected_items_per_pipeline
size_t	m_expected_total_channels = 0
std::mutex	m_results_mutex
ResultContainer	m_latest_results
std::thread	m_collection_thread
std::atomic< bool >	m_thread_should_stop {false}
std::atomic< bool >	m_thread_running {false}
std::atomic< bool >	m_harvest_requested {false}
std::mutex	m_config_mutex
std::mutex	m_collection_mutex
std::condition_variable	m_collection_cv

Detailed Description

Definition at line 29 of file TPGInternalStateHarvester.hpp.

Member Typedef Documentation

ProcRef

using dunedaq::fdreadoutlibs::TPGInternalStateHarvester::ProcRef = std::pair<std::shared_ptr<tpglibs::AbstractProcessor<__m256i>>, int >

Definition at line 31 of file TPGInternalStateHarvester.hpp.

ResultContainer

using dunedaq::fdreadoutlibs::TPGInternalStateHarvester::ResultContainer

private

Initial value:

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

Definition at line 109 of file TPGInternalStateHarvester.hpp.

Constructor & Destructor Documentation

~TPGInternalStateHarvester()

dunedaq::fdreadoutlibs::TPGInternalStateHarvester::~TPGInternalStateHarvester

(

)

Definition at line 20 of file TPGInternalStateHarvester.cpp.

                                                      {
  // Ensure thread is properly stopped
  stop_collection_thread();
}

Member Function Documentation

collection_thread_worker_()

void dunedaq::fdreadoutlibs::TPGInternalStateHarvester::collection_thread_worker_ ( )

private

Definition at line 250 of file TPGInternalStateHarvester.cpp.

                                                          {
  while (!m_thread_should_stop.load()) {
    std::unique_lock<std::mutex> lock(m_collection_mutex);
    
    // Wait for harvest request or stop signal
    m_collection_cv.wait(lock, [this] {
      return m_harvest_requested.load() || m_thread_should_stop.load();
    });
    
    if (m_thread_should_stop.load()) {
      break;
    }
    
    // Reset the harvest request flag
    m_harvest_requested.store(false);
    
    // Release lock during collection to allow concurrent reads
    lock.unlock();
    
    // Perform the actual harvest (expensive operation in background)
    auto new_results = harvest_once();
    
    // Update results with simple mutex protection
    {
      std::lock_guard<std::mutex> results_lock(m_results_mutex);
      m_latest_results = std::move(new_results);
    }
  }
}

get_latest_results()

std::unordered_map< trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > > dunedaq::fdreadoutlibs::TPGInternalStateHarvester::get_latest_results

(

)

const

Get the latest collected results (thread-safe, non-blocking) Returns a copy of the most recent harvest results.

Returns

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

Definition at line 241 of file TPGInternalStateHarvester.cpp.

                                                    {
  std::lock_guard<std::mutex> lock(m_results_mutex);
  return m_latest_results; // Return copy under lock (blocking read is acceptable)
}

get_processor_references()

const std::vector< TPGInternalStateHarvester::ProcRef > & dunedaq::fdreadoutlibs::TPGInternalStateHarvester::get_processor_references

(

)

const

Definition at line 31 of file TPGInternalStateHarvester.cpp.

                                                          {
  return m_processor_references;
}

harvest_once()

std::unordered_map< trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > > dunedaq::fdreadoutlibs::TPGInternalStateHarvester::harvest_once

(

)

Harvest once, outputs channel -> [(metric_name, value)...].

Returns

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

Definition at line 85 of file TPGInternalStateHarvester.cpp.

{
  std::unordered_map<trgdataformats::channel_t,
                     std::vector<std::pair<std::string,int16_t>>> out;
 
  // 1. pre-estimate the capacity of the buckets (usually 64)
  if (m_expected_total_channels > 0) {
    out.reserve(m_expected_total_channels);
  } else {
    out.reserve(static_cast<size_t>(m_num_channels_per_pipeline) * m_num_pipelines);
  }
 
  // defensive: if the cache is not complete for the current pipelines, rebuild it
  if (m_expected_items_per_pipeline.size() != m_num_pipelines) {
    rebuild_prealloc_caches_();
  }
  
  for (size_t i = 0; i < m_processor_references.size(); ++i) {
    const auto& [proc, pipeline_id] = m_processor_references[i];
    if (!proc) {
      continue;
    }
 
    // 1) get the cached metric names; if empty, fall back to pulling directly from processor
    std::vector<std::string> metric_names_cached;
    if (m_metric_items_per_proc.size() > i) {
      metric_names_cached = m_metric_items_per_proc[i];
    } else {
      // Fallback: use processor's interface method
      metric_names_cached = proc->get_requested_internal_state_names();
    }
 
    // 2) get the current snapshot
    const auto arr = proc->read_internal_states_as_integer_array();
 
    // basic consistency: the number of items should be the same as the number of snapshot items
    if (metric_names_cached.size() != arr.m_size) {
      TLOG_DEBUG(TLVL_BOOKKEEPING) << "Processor " << i << " size mismatch: metric_names=" 
                                   << metric_names_cached.size() << " vs array=" << arr.m_size;
      continue;
    }
 
    // the current pipeline's lane -> (channel, plane)
    assert(static_cast<size_t>(pipeline_id) < m_channel_plane_numbers_per_pipeline.size());
    const auto& chan_plane_vec = m_channel_plane_numbers_per_pipeline[pipeline_id];
 
    // 3) first reserve the capacity for all channels in the current pipeline in out
    //    so that subsequent emplace_back will not trigger allocation
    std::vector<decltype(out.begin())> iters_for_lanes;
    iters_for_lanes.resize(chan_plane_vec.size());
 
    const size_t expected_items_here =
      (pipeline_id < m_expected_items_per_pipeline.size())
      ? m_expected_items_per_pipeline[pipeline_id]
      : metric_names_cached.size(); // fallback, use the number of items of the current processor
 
    for (size_t lane = 0; lane < chan_plane_vec.size(); ++lane) {
      const auto ch = chan_plane_vec[lane].first; // offline channel id
      auto [it, inserted] = out.try_emplace(ch, std::vector<std::pair<std::string,int16_t>>{});
      if (inserted) {
        // only reserve the capacity for the first time the channel is encountered
        it->second.reserve(expected_items_here);
      }
      iters_for_lanes[lane] = it;
    }
 
    // 4) map the 16-lane array of each metric to the corresponding channel
    for (size_t item = 0; item < arr.m_size; ++item) {
      const auto& lanes = arr.m_data[item]; // std::array<int16_t, 16>
      const auto& name  = metric_names_cached[item];
      const size_t L    = lanes.size();     // e.g. 16
 
      const size_t up_to = std::min(L, chan_plane_vec.size());
      for (size_t lane = 0; lane < up_to; ++lane) {
        // directly use the iterator, avoid the repeated lookup of the map
        iters_for_lanes[lane]->second.emplace_back(name, lanes[lane]);
      }
    }
  }
 
  return out;
}

is_collection_thread_running()

bool dunedaq::fdreadoutlibs::TPGInternalStateHarvester::is_collection_thread_running

(

)

const

Check if collection thread is running.

Returns

true if thread is active, false otherwise

Definition at line 246 of file TPGInternalStateHarvester.cpp.

                                                                   {
  return m_thread_running.load();
}

rebuild_prealloc_caches_()

void dunedaq::fdreadoutlibs::TPGInternalStateHarvester::rebuild_prealloc_caches_ ( )

private

Definition at line 57 of file TPGInternalStateHarvester.cpp.

{
  // pre-clean
  m_metric_items_per_proc.clear();
  m_metric_items_per_proc.reserve(m_processor_references.size());
 
  // first clear/reset the expected number of metric items per pipeline
  m_expected_items_per_pipeline.clear();
  m_expected_items_per_pipeline.resize(m_num_pipelines, 0);
 
  // cache the metric names for each processor and accumulate to the corresponding pipeline
  for (const auto& [proc, pipeline_id] : m_processor_references) {
    if (proc) {
      // Use the processor's interface method which delegates to the registry
      auto names = proc->get_requested_internal_state_names();
      if (static_cast<size_t>(pipeline_id) < m_expected_items_per_pipeline.size()) {
        m_expected_items_per_pipeline[pipeline_id] += names.size();
      }
      m_metric_items_per_proc.emplace_back(std::move(names));
    } else {
      m_metric_items_per_proc.emplace_back(); // empty
    }
  }
}

set_processor_references()

void dunedaq::fdreadoutlibs::TPGInternalStateHarvester::set_processor_references

(

std::vector< ProcRef >

refs

)

Definition at line 25 of file TPGInternalStateHarvester.cpp.

                                                                                {
  m_processor_references = std::move(refs);
  rebuild_prealloc_caches_(); // if num_pipelines is not set, this will gracefully get empty per-pipeline statistics, which will be filled later by update
}

start_collection_thread()

void dunedaq::fdreadoutlibs::TPGInternalStateHarvester::start_collection_thread

(

)

Start the background collection thread Must be called before using async collection features.

Definition at line 170 of file TPGInternalStateHarvester.cpp.

                                                        {
  std::lock_guard<std::mutex> lock(m_config_mutex);
  
  if (m_thread_running.load()) {
    return; // Already running
  }
  
  TLOG_DEBUG(TLVL_BOOKKEEPING) << "Starting internal state collection thread";
  
  // Initialize result container
  m_latest_results.clear();
  
  // Reset thread control flags
  m_thread_should_stop.store(false);
  m_harvest_requested.store(false);
  
  // Start the collection thread
  m_collection_thread = std::thread(&TPGInternalStateHarvester::collection_thread_worker_, this);
  m_thread_running.store(true);
}

stop_collection_thread()

void dunedaq::fdreadoutlibs::TPGInternalStateHarvester::stop_collection_thread

(

)

Stop the background collection thread Blocks until thread is fully stopped.

Definition at line 191 of file TPGInternalStateHarvester.cpp.

                                                       {
  {
    std::lock_guard<std::mutex> config_lock(m_config_mutex);
    
    if (!m_thread_running.load()) {
      return; // Already stopped
    }
    
    TLOG_DEBUG(TLVL_BOOKKEEPING) << "Stopping internal state collection thread";
    
    // Signal thread to stop
    m_thread_should_stop.store(true);
  }
  
  // Notify the collection thread using the correct mutex
  {
    std::lock_guard<std::mutex> collection_lock(m_collection_mutex);
    m_collection_cv.notify_all();
  }
  
  // Wait for thread to finish
  if (m_collection_thread.joinable()) {
    m_collection_thread.join();
  }
  
  m_thread_running.store(false);
  
  // Clear results
  {
    std::lock_guard<std::mutex> lock(m_results_mutex);
    m_latest_results.clear();
  }
}

trigger_harvest()

void dunedaq::fdreadoutlibs::TPGInternalStateHarvester::trigger_harvest

(

)

Signal the collection thread to perform one harvest cycle Non-blocking - returns immediately.

Definition at line 225 of file TPGInternalStateHarvester.cpp.

                                                {
  if (!m_thread_running.load()) {
    return; // Thread not running
  }
  
  m_harvest_requested.store(true);
  
  // Notify the collection thread using the correct mutex
  {
    std::lock_guard<std::mutex> lock(m_collection_mutex);
    m_collection_cv.notify_all();
  }
}

update_channel_plane_numbers()

void dunedaq::fdreadoutlibs::TPGInternalStateHarvester::update_channel_plane_numbers	(	const std::vector< std::pair< trgdataformats::channel_t, int16_t > > &	channel_plane_numbers,
		uint8_t	num_channels_per_pipeline,
		uint8_t	num_pipelines )

Cuts a full list of (channel, plane) into per-pipeline lists of 16 lanes each.

Parameters

channel_plane_numbers
num_channels_per_pipeline
num_pipelines

Definition at line 35 of file TPGInternalStateHarvester.cpp.

{
  m_channel_plane_numbers_per_pipeline.clear();
  m_channel_plane_numbers_per_pipeline.resize(num_pipelines);
 
  // Cut in order: each pipeline has num_channels_per_pipeline channels
  for (uint8_t p = 0; p < num_pipelines; ++p) {
    auto begin = channel_plane_numbers.begin() + p * num_channels_per_pipeline;
    auto end   = begin + num_channels_per_pipeline;
    m_channel_plane_numbers_per_pipeline[p] =
      std::vector<std::pair<trgdataformats::channel_t,int16_t>>(begin, end);
  }
  m_num_channels_per_pipeline = num_channels_per_pipeline;
  m_num_pipelines = num_pipelines;
  m_expected_total_channels = static_cast<size_t>(m_num_channels_per_pipeline) * m_num_pipelines;
 
  rebuild_prealloc_caches_(); // now pipelines number is known, can accurately calculate the expected number of metric items per pipeline
}

Member Data Documentation

m_channel_plane_numbers_per_pipeline

std::vector<std::vector<std::pair<trgdataformats::channel_t,int16_t> > > dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_channel_plane_numbers_per_pipeline

private

Definition at line 99 of file TPGInternalStateHarvester.hpp.

m_collection_cv

std::condition_variable dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_collection_cv

private

Definition at line 125 of file TPGInternalStateHarvester.hpp.

m_collection_mutex

std::mutex dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_collection_mutex

private

Definition at line 124 of file TPGInternalStateHarvester.hpp.

m_collection_thread

std::thread dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_collection_thread

private

Definition at line 117 of file TPGInternalStateHarvester.hpp.

m_config_mutex

std::mutex dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_config_mutex

mutableprivate

Definition at line 123 of file TPGInternalStateHarvester.hpp.

m_expected_items_per_pipeline

std::vector<size_t> dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_expected_items_per_pipeline

private

Definition at line 105 of file TPGInternalStateHarvester.hpp.

m_expected_total_channels

size_t dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_expected_total_channels = 0

private

Definition at line 106 of file TPGInternalStateHarvester.hpp.

m_harvest_requested

std::atomic<bool> dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_harvest_requested {false}

private

Definition at line 120 of file TPGInternalStateHarvester.hpp.

{false};

m_latest_results

ResultContainer dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_latest_results

private

Definition at line 114 of file TPGInternalStateHarvester.hpp.

m_metric_items_per_proc

std::vector<std::vector<std::string> > dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_metric_items_per_proc

private

Definition at line 104 of file TPGInternalStateHarvester.hpp.

m_num_channels_per_pipeline

uint8_t dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_num_channels_per_pipeline

private

Definition at line 100 of file TPGInternalStateHarvester.hpp.

m_num_pipelines

uint8_t dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_num_pipelines

private

Definition at line 101 of file TPGInternalStateHarvester.hpp.

m_processor_references

std::vector<ProcRef> dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_processor_references

private

Definition at line 97 of file TPGInternalStateHarvester.hpp.

m_results_mutex

std::mutex dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_results_mutex

mutableprivate

Definition at line 113 of file TPGInternalStateHarvester.hpp.

m_thread_running

std::atomic<bool> dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_thread_running {false}

private

Definition at line 119 of file TPGInternalStateHarvester.hpp.

{false};

m_thread_should_stop

std::atomic<bool> dunedaq::fdreadoutlibs::TPGInternalStateHarvester::m_thread_should_stop {false}

private

Definition at line 118 of file TPGInternalStateHarvester.hpp.

{false};

---

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

• /github/workspace/dunedaq/sourcecode/fdreadoutlibs/include/fdreadoutlibs/tpg/TPGInternalStateHarvester.hpp
• /github/workspace/dunedaq/sourcecode/fdreadoutlibs/src/tpg/TPGInternalStateHarvester.cpp