TAEmulationWorker.hpp

Go to the documentation of this file.

#ifndef TRGTOOLS_TAEMULATIONWORKER_HPP_
#define TRGTOOLS_TAEMULATIONWORKER_HPP_
 
#include "trgtools/TAEmulationUnit.hpp"
 
#include "CLI/App.hpp"
#include "CLI/Config.hpp"
#include "CLI/Formatter.hpp"
 
#include <filesystem>
#include <fmt/chrono.h>
#include <fmt/core.h>
#include <fmt/format.h>
 
#include "hdf5libs/HDF5RawDataFile.hpp"
#include "trgdataformats/TriggerPrimitive.hpp"
#include "triggeralgs/TriggerActivityFactory.hpp"
#include "triggeralgs/TriggerObjectOverlay.hpp"
 
namespace dunedaq::trgtools {
 
class TAEmulationWorker
{
public:
  TAEmulationWorker(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);
 
  ~TAEmulationWorker() = default;

  std::vector<daqdataformats::SourceID> get_valid_sourceids(daqdataformats::TimeSlice& _timeslice);

  void start_processing();

  void wait_to_complete_work();

  std::map<uint64_t, std::vector<triggeralgs::TriggerActivity>> get_tas();

  std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>> get_frags();

  hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t get_sourceid_geoid_map();
 
private:
  void process_tasks();

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

  void worker_thread();

  void enqueue_task(std::function<void()> task);

  void wait_to_complete_tasks();
 
private:
  std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>> m_input_files;

  nlohmann::json m_configuration;

  std::vector<std::string> m_input_paths;

  std::map<daqdataformats::SourceID, std::unique_ptr<trgtools::TAEmulationUnit>> m_ta_emulators;

  std::pair<uint64_t, uint64_t> m_sliceid_range;

  const bool m_run_parallel;

  const bool m_quiet;
 
  /*
   * Threading objects for the main file handler
   */

  std::thread m_main_thread;

  std::atomic<bool> m_stop{ false };
 
  /*
   * Optional threading objects for the tasks
   * i.e. one thread per TAMaker
   */

  std::mutex m_savetps_mutex;
  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;

  std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>> m_ta_fragments;

  uint16_t m_id;
  static uint16_t m_id_next;
  static const size_t SIZE_TP = sizeof(trgdataformats::TriggerPrimitive);
};
 
};
 
#endif