dunedaq::trgtools::TAFileHandler Class Reference

#include <TAFileHandler.hpp>

Collaboration diagram for dunedaq::trgtools::TAFileHandler:

Public Member Functions
	TAFileHandler (std::vector< std::shared_ptr< hdf5libs::HDF5RawDataFile > > _input_files, nlohmann::json _config, std::pair< uint64_t, uint64_t > _sliceid_range, bool _run_parallel, bool _quiet)
	Constructor, takes file input path & configuration.
	~TAFileHandler ()=default
std::vector< daqdataformats::SourceID >	get_valid_sourceids (daqdataformats::TimeSlice &_timeslice)
	Get the valid sourceids object from HDF5 file.
void	start_processing ()
	User interaction for task processing.
void	wait_to_complete_work ()
	Waits for all the tasks to complete.
std::map< uint64_t, std::vector< triggeralgs::TriggerActivity > >	get_tas ()
	Retrieves all the TAs with std::move operator.
std::map< uint64_t, std::vector< std::unique_ptr< daqdataformats::Fragment > > >	get_frags ()
	Retrieves all the unique pointers to the TA fragments.
hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t	get_sourceid_geoid_map ()
	Get the sourceid to geoid map object.

Private Member Functions
void	process_tasks ()
	Function that processes the whole file.
void	process_task (daqdataformats::SourceID _source_id, uint64_t _rec, daqdataformats::FragmentHeader _header, std::vector< trgdataformats::TriggerPrimitive > &&_tps)
	Function that processes one slice for one plane.
void	worker_thread ()
	Creates & runs a worker thread.
void	enqueue_task (std::function< void()> task)
	Enqueues task to process.
void	wait_to_complete_tasks ()
	Waits to complete a task.

Private Attributes
std::vector< std::shared_ptr< hdf5libs::HDF5RawDataFile > >	m_input_files
	A pointer to the input file.
nlohmann::json	m_configuration
	configuration for the TA-makers
std::vector< std::string >	m_input_paths
	input vector of tpstream input paths
std::map< daqdataformats::SourceID, std::unique_ptr< trgtools::EmulateTAUnit > >	m_ta_emulators
	Map of SourceID : Emulator unit (TAMaker)
std::pair< uint64_t, uint64_t >	m_sliceid_range
	Range of TimeSlice IDs to process.
const bool	m_run_parallel
	Run the TA makers in parllel.
const bool	m_quiet
	Quiet down the cout output.
std::thread	m_main_thread
	The file handler thread.
std::atomic< bool >	m_stop {false}
	Bool to indicate to stop the emulation.
std::mutex	m_savetps_mutex
	Mutex for saving the TPs.
std::vector< std::thread >	m_thread_pool
std::condition_variable	m_condition
std::condition_variable	m_task_complete_condition
std::queue< std::function< void()> >	m_task_queue
std::mutex	m_queue_mutex
std::atomic< size_t >	m_active_tasks
std::map< uint64_t, std::vector< triggeralgs::TriggerActivity > >	m_tas
	Output vector of TAs.
std::map< uint64_t, std::vector< std::unique_ptr< daqdataformats::Fragment > > >	m_ta_fragments
	Output vector of TA fragments.
uint16_t	m_id
	Unique ID for this TAFileHandler.

Static Private Attributes
static uint16_t	m_id_next = 0
	Global variable used to get the next ID.
static const size_t	SIZE_TP = sizeof(trgdataformats::TriggerPrimitive)
	Size of the TPS.

Detailed Description

Definition at line 23 of file TAFileHandler.hpp.

Constructor & Destructor Documentation

TAFileHandler()

dunedaq::trgtools::TAFileHandler::TAFileHandler	(	std::vector< std::shared_ptr< hdf5libs::HDF5RawDataFile > >	_input_files,
		nlohmann::json	_config,
		std::pair< uint64_t, uint64_t >	_sliceid_range,
		bool	_run_parallel,
		bool	_quiet )

Constructor, takes file input path & configuration.

Each TAFileHandler will create its own thread, so all TAFileHandlers are run on separate threads

Parameters

_input_files	a vector of input HDF5 shared pointers to process
_config	TAMaker configuration
_sliceid_range	range of sliceids to process
_run_parallel	run each TAMaker (one per SourceID) in parallel
_quiet	quiet down the cout

Definition at line 11 of file TAFileHandler.cpp.

  : m_input_files(input_files),
    m_sliceid_range(sliceid_range),
    m_run_parallel(run_parallel),
    m_quiet(quiet),
    m_id(m_id_next++)
{
  std::string algo_name = config["trigger_activity_plugin"][0];
  nlohmann::json algo_config = config["trigger_activity_config"][0];
 
  // Get the input file
  // Extract the run number etc
  std::vector<daqdataformats::run_number_t> run_numbers;
  std::vector<size_t> file_indices;
  for (const auto& input_file : input_files) {
    if (std::find(run_numbers.begin(), run_numbers.end(),
        input_file->get_attribute<daqdataformats::run_number_t>("run_number")) ==
        run_numbers.end()) {
          run_numbers.push_back(input_file->get_attribute<daqdataformats::run_number_t>("run_number"));
    }
 
    if (std::find(file_indices.begin(), file_indices.end(),
        input_file->get_attribute<daqdataformats::run_number_t>("run_number")) ==
        file_indices.end()) {
          file_indices.push_back(input_file->get_attribute<size_t>("file_index"));
    }
  }
 
  std::string application_name = m_input_files.front()->get_attribute<std::string>("application_name");
 
  if (!m_quiet) {
    fmt::print("Run Numbers: {}\nFile Indices: {}\nApp name: '{}'\n", fmt::join(run_numbers, ","), fmt::join(file_indices, ","), application_name);
  }
 
  // std::set of record IDs (pair of record number & sequence number)
  auto records = m_input_files.front()->get_all_record_ids();
 
  // Extract the number of TAMakers to create
  daqdataformats::TimeSlice first_timeslice = m_input_files.front()->get_timeslice(*records.begin());
  std::vector<daqdataformats::SourceID> valid_sources = get_valid_sourceids(first_timeslice);
  fmt::print("Number of makers to make: {}\n", valid_sources.size());
 
  for (const daqdataformats::SourceID& sid : valid_sources) {
    // Create TAMaker
    std::unique_ptr<triggeralgs::TriggerActivityMaker> ta_maker =
      triggeralgs::TriggerActivityFactory::get_instance()->build_maker(algo_name);
    ta_maker->configure(algo_config);
 
    // Add it to the enulators
    m_ta_emulators[sid] = std::make_unique<trgtools::EmulateTAUnit>();
    m_ta_emulators[sid]->set_maker(ta_maker);
 
    // Create a worker thread per emulator
    if (m_run_parallel) {
      m_thread_pool.emplace_back(&TAFileHandler::worker_thread, this);
    }
  }
}

~TAFileHandler()

dunedaq::trgtools::TAFileHandler::~TAFileHandler ( )

default

Member Function Documentation

enqueue_task()

void dunedaq::trgtools::TAFileHandler::enqueue_task ( std::function< void()> task )

private

Enqueues task to process.

Parameters

task	task to porcess

Definition at line 219 of file TAFileHandler.cpp.

{
  {
    std::lock_guard<std::mutex> lock(m_queue_mutex);
    m_task_queue.push(std::move(task));
    ++m_active_tasks;
  }
  m_condition.notify_one();
}

get_frags()

std::map< uint64_t, std::vector< std::unique_ptr< daqdataformats::Fragment > > > dunedaq::trgtools::TAFileHandler::get_frags

(

)

Retrieves all the unique pointers to the TA fragments.

Returns

A map of sourceID : TA fragment

Definition at line 304 of file TAFileHandler.cpp.

{
  return std::move(m_ta_fragments);
}

get_sourceid_geoid_map()

hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t dunedaq::trgtools::TAFileHandler::get_sourceid_geoid_map

(

)

Get the sourceid to geoid map object.

Returns

geoid to sourceid map

Definition at line 94 of file TAFileHandler.cpp.

{
  if (!m_input_files.size()) {
    throw "Files not set yet!";
  }
 
  return m_input_files.front()->get_srcid_geoid_map();
}

get_tas()

std::map< uint64_t, std::vector< triggeralgs::TriggerActivity > > dunedaq::trgtools::TAFileHandler::get_tas

(

)

Retrieves all the TAs with std::move operator.

Returns

a map of sourceID : TA

Definition at line 299 of file TAFileHandler.cpp.

{
  return std::move(m_tas);
}

get_valid_sourceids()

std::vector< daqdataformats::SourceID > dunedaq::trgtools::TAFileHandler::get_valid_sourceids

(

daqdataformats::TimeSlice &

_timeslice

)

Get the valid sourceids object from HDF5 file.

Parameters

_timeslice

timeslice to load the sourceIDs from

Returns

vector of SoureIDs from this file

Definition at line 75 of file TAFileHandler.cpp.

{
  const auto& fragments = _timeslice.get_fragments_ref();
 
  std::vector<daqdataformats::SourceID> ret;
  for (const auto& fragment : fragments) {
    if (fragment->get_fragment_type() != daqdataformats::FragmentType::kTriggerPrimitive) {
      continue;
    }
 
    daqdataformats::SourceID sourceid = fragment->get_element_id();
 
    ret.push_back(sourceid);
  }
 
  return ret;
}

process_task()

void dunedaq::trgtools::TAFileHandler::process_task ( daqdataformats::SourceID _source_id, uint64_t _rec, daqdataformats::FragmentHeader _header, std::vector< trgdataformats::TriggerPrimitive > && _tps )

private

Function that processes one slice for one plane.

Parameters

_source_id	sourceID to process
_rec	record ID to process
_header	fragment header
_tps	vectors of trigger primitives to process (with move operator)

Definition at line 258 of file TAFileHandler.cpp.

{
  // Get te last fragment
  std::unique_ptr<daqdataformats::Fragment> frag = m_ta_emulators[_source_id]->emulate_vector(_tps);
 
  // Don't do anything if no fragments found
  if (!frag) {
    return;
  }
 
  // Get all the TriggerActivities from the TA Emulator buffer
  std::vector<triggeralgs::TriggerActivity> ta_buffer = m_ta_emulators[_source_id]->get_last_output_buffer();
 
  // Don't continue if no TAs found
  size_t n_tas = ta_buffer.size();
  if (!n_tas) {
    return;
  }
 
  if (!m_quiet && n_tas) {
    fmt::print(" Found {} TAs!\n", n_tas);
  }
 
  // Set the fragment header & push into our output (with locking!)
  {
    if (m_run_parallel) {
      std::lock_guard<std::mutex> lock(m_savetps_mutex);
    }
    m_tas[_rec].reserve(m_tas[_rec].size() + ta_buffer.size());
    m_tas[_rec].insert(m_tas[_rec].end(), std::make_move_iterator(ta_buffer.begin()), std::make_move_iterator(ta_buffer.end()));
 
    frag->set_header_fields(_header);
    frag->set_type(daqdataformats::FragmentType::kTriggerActivity);
 
    m_ta_fragments[_rec].push_back(std::move(frag));
  }
}

process_tasks()

void dunedaq::trgtools::TAFileHandler::process_tasks ( )

private

Function that processes the whole file.

Definition at line 134 of file TAFileHandler.cpp.

{
  // Iterate over the input files
  for (auto& input_file: m_input_files) {
    // std::set of record IDs (pair of record number & sequence number)
    auto records = input_file->get_all_record_ids();
 
    for (const auto& record : records) {
      if (record.first < m_sliceid_range.first || record.first > m_sliceid_range.second) {
        if (!m_quiet)
          fmt::print("  Will not process RecordID {} because it's outside of our range!", record.first);
        continue;
      }
 
      // Get all the fragments
      daqdataformats::TimeSlice timeslice = input_file->get_timeslice(record);
      const auto& fragments = timeslice.get_fragments_ref();
 
      // Iterate over the fragments & process each fragment
      for (const auto& fragment : fragments) {
        daqdataformats::SourceID sid = fragment->get_element_id();
 
        if (!m_ta_emulators.contains(sid)) {
          continue;
        }
 
        // Pull tps out
        size_t n_tps = fragment->get_data_size()/SIZE_TP;
        if (!m_quiet) {
          fmt::print("  TP fragment size: {}\n", fragment->get_data_size());
          fmt::print("  Num TPs: {}\n", n_tps);
        }
 
        // Create a TP buffer
        std::vector<trgdataformats::TriggerPrimitive> tp_buffer;
        // Prepare the TP buffer, checking for time ordering
        tp_buffer.reserve(n_tps);
 
        // Populate the TP buffer
        trgdataformats::TriggerPrimitive* tp_array = static_cast<trgdataformats::TriggerPrimitive*>(fragment->get_data());
        uint64_t last_ts = 0;
        for(size_t tpid(0); tpid<n_tps; ++tpid) {
          auto& tp = tp_array[tpid];
          if (tp.time_start <= last_ts && !m_quiet) {
            fmt::print("  ERROR: {} {} ", +tp.time_start, last_ts );
          }
          tp_buffer.push_back(tp);
        }
 
        daqdataformats::FragmentHeader frag_hdr = fragment->get_header();
 
        // Customise the source id (add 1000 to id)
        frag_hdr.element_id = daqdataformats::SourceID{daqdataformats::SourceID::Subsystem::kTrigger, fragment->get_element_id().id+1000};
 
        // Either enqueue the task if using parallel processing, or execute the task now
        if (m_run_parallel) {
          enqueue_task([this, sid, record, frag_hdr, tp_buffer = std::move(tp_buffer)]() mutable {
            this->process_task(sid, record.first, frag_hdr, std::move(tp_buffer));
          });
        }
        else {
          this->process_task(sid, record.first, frag_hdr, std::move(tp_buffer));
        }
      }
      // If running in parallel, wait to process entire slice before we move to
      // the next one
      if (m_run_parallel) {
        wait_to_complete_tasks();
      }
    }
 
    size_t total = 0;
    for (auto& [key, vec_tas]: m_tas) {
      total += vec_tas.size();
    }
    std::cout << "We have a total of " << total << " TAs!" << std::endl;
  }
}

start_processing()

void dunedaq::trgtools::TAFileHandler::start_processing

(

)

User interaction for task processing.

Definition at line 213 of file TAFileHandler.cpp.

{
  m_main_thread = std::thread(&TAFileHandler::process_tasks, this);
}

wait_to_complete_tasks()

void dunedaq::trgtools::TAFileHandler::wait_to_complete_tasks ( )

private

Waits to complete a task.

Definition at line 229 of file TAFileHandler.cpp.

{
  std::unique_lock<std::mutex> lock(m_queue_mutex);
  m_task_complete_condition.wait(lock, [this]() { return m_active_tasks == 0; });
}

wait_to_complete_work()

void dunedaq::trgtools::TAFileHandler::wait_to_complete_work

(

)

Waits for all the tasks to complete.

Definition at line 235 of file TAFileHandler.cpp.

{
  // Wait for the main threads to join
  m_main_thread.join();
  fmt::print("TAFileHandler_{} work completed\n", m_id);
 
  // Wait for the tasks to complete
  if (m_run_parallel) {
    wait_to_complete_tasks();
 
    {
      std::lock_guard<std::mutex> lock(m_queue_mutex);
      m_stop = true;
    }
    m_condition.notify_all();
 
    fmt::print("m_stop issued\n");
    for (std::thread& thread : m_thread_pool) {
      thread.join();
    }
  }
}

worker_thread()

void dunedaq::trgtools::TAFileHandler::worker_thread ( )

private

Creates & runs a worker thread.

Get a task from the queue (with locking)

Definition at line 103 of file TAFileHandler.cpp.

{
  while (true) {
    std::function<void()> task;
    {
      std::unique_lock<std::mutex> lock(m_queue_mutex);
      m_condition.wait(lock, [this]() {return m_stop || !m_task_queue.empty(); });
 
      if (m_stop && m_task_queue.empty()) {
        return;
      }
 
      task = std::move(m_task_queue.front());
      m_task_queue.pop();
    }
 
    // Run & complete a task
    task();
 
    // Notify that task was completed
    {
      std::lock_guard<std::mutex> lock(m_queue_mutex);
      --m_active_tasks;
      if (m_active_tasks == 0) {
        m_task_complete_condition.notify_all();
      }
    }
  }
}

Member Data Documentation

m_active_tasks

std::atomic<size_t> dunedaq::trgtools::TAFileHandler::m_active_tasks

private

Definition at line 160 of file TAFileHandler.hpp.

m_condition

std::condition_variable dunedaq::trgtools::TAFileHandler::m_condition

private

Definition at line 156 of file TAFileHandler.hpp.

m_configuration

nlohmann::json dunedaq::trgtools::TAFileHandler::m_configuration

private

configuration for the TA-makers

Definition at line 121 of file TAFileHandler.hpp.

m_id

uint16_t dunedaq::trgtools::TAFileHandler::m_id

private

Unique ID for this TAFileHandler.

Definition at line 169 of file TAFileHandler.hpp.

m_id_next

uint16_t dunedaq::trgtools::TAFileHandler::m_id_next = 0

staticprivate

Global variable used to get the next ID.

Definition at line 171 of file TAFileHandler.hpp.

m_input_files

std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile> > dunedaq::trgtools::TAFileHandler::m_input_files

private

A pointer to the input file.

Definition at line 118 of file TAFileHandler.hpp.

m_input_paths

std::vector<std::string> dunedaq::trgtools::TAFileHandler::m_input_paths

private

input vector of tpstream input paths

Definition at line 124 of file TAFileHandler.hpp.

m_main_thread

std::thread dunedaq::trgtools::TAFileHandler::m_main_thread

private

The file handler thread.

Definition at line 143 of file TAFileHandler.hpp.

m_queue_mutex

std::mutex dunedaq::trgtools::TAFileHandler::m_queue_mutex

private

Definition at line 159 of file TAFileHandler.hpp.

m_quiet

const bool dunedaq::trgtools::TAFileHandler::m_quiet

private

Quiet down the cout output.

Definition at line 136 of file TAFileHandler.hpp.

m_run_parallel

const bool dunedaq::trgtools::TAFileHandler::m_run_parallel

private

Run the TA makers in parllel.

Definition at line 133 of file TAFileHandler.hpp.

m_savetps_mutex

std::mutex dunedaq::trgtools::TAFileHandler::m_savetps_mutex

private

Mutex for saving the TPs.

Definition at line 154 of file TAFileHandler.hpp.

m_sliceid_range

std::pair<uint64_t, uint64_t> dunedaq::trgtools::TAFileHandler::m_sliceid_range

private

Range of TimeSlice IDs to process.

Definition at line 130 of file TAFileHandler.hpp.

m_stop

std::atomic<bool> dunedaq::trgtools::TAFileHandler::m_stop {false}

private

Bool to indicate to stop the emulation.

Definition at line 146 of file TAFileHandler.hpp.

{false};

m_ta_emulators

std::map<daqdataformats::SourceID, std::unique_ptr<trgtools::EmulateTAUnit> > dunedaq::trgtools::TAFileHandler::m_ta_emulators

private

Map of SourceID : Emulator unit (TAMaker)

Definition at line 127 of file TAFileHandler.hpp.

m_ta_fragments

std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment> > > dunedaq::trgtools::TAFileHandler::m_ta_fragments

private

Output vector of TA fragments.

Definition at line 166 of file TAFileHandler.hpp.

m_tas

std::map<uint64_t, std::vector<triggeralgs::TriggerActivity> > dunedaq::trgtools::TAFileHandler::m_tas

private

Output vector of TAs.

Definition at line 163 of file TAFileHandler.hpp.

m_task_complete_condition

std::condition_variable dunedaq::trgtools::TAFileHandler::m_task_complete_condition

private

Definition at line 157 of file TAFileHandler.hpp.

m_task_queue

std::queue<std::function<void()> > dunedaq::trgtools::TAFileHandler::m_task_queue

private

Definition at line 158 of file TAFileHandler.hpp.

m_thread_pool

std::vector<std::thread> dunedaq::trgtools::TAFileHandler::m_thread_pool

private

Definition at line 155 of file TAFileHandler.hpp.

SIZE_TP

const size_t dunedaq::trgtools::TAFileHandler::SIZE_TP = sizeof(trgdataformats::TriggerPrimitive)

staticprivate

Size of the TPS.

Definition at line 173 of file TAFileHandler.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/trgtools/include/trgtools/TAFileHandler.hpp
• /github/workspace/dunedaq/sourcecode/trgtools/src/TAFileHandler.cpp