dunedaq::snbmodules::FileSourceModel< ReadoutType > Class Template Reference

#include <FileSourceModel.hpp>

Inheritance diagram for dunedaq::snbmodules::FileSourceModel< ReadoutType >:

Collaboration diagram for dunedaq::snbmodules::FileSourceModel< ReadoutType >:

Public Member Functions
	FileSourceModel (std::string name, std::atomic< bool > &run_marker, double rate_khz)
void	set_sender (const std::string &conn_name)
void	conf (const confmodel::DetectorStream *stream_conf, const appmodel::SNBFileSourceParameters *file_params)
void	scrap (const appfwk::DAQModule::CommandData_t &)
bool	is_configured () override
void	start (const appfwk::DAQModule::CommandData_t &)
void	stop (const appfwk::DAQModule::CommandData_t &)
Public Member Functions inherited from dunedaq::snbmodules::FileSourceConcept
	FileSourceConcept ()
virtual	~FileSourceConcept ()
	FileSourceConcept (const FileSourceConcept &)=delete
	FileSourceConcept is not copy-constructible.
FileSourceConcept &	operator= (const FileSourceConcept &)=delete
	FileSourceConcept is not copy-assginable.
	FileSourceConcept (FileSourceConcept &&)=delete
	FileSourceConcept is not move-constructible.
FileSourceConcept &	operator= (FileSourceConcept &&)=delete
	FileSourceConcept is not move-assignable.
Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject (const MonitorableObject &)=delete
MonitorableObject &	operator= (const MonitorableObject &)=delete
	MonitorableObject (MonitorableObject &&)=delete
MonitorableObject &	operator= (MonitorableObject &&)=delete
virtual	~MonitorableObject ()=default
auto	get_opmon_id () const noexcept
auto	get_opmon_level () const noexcept

Protected Member Functions
void	run_produce ()
virtual void	generate_opmon_data () override
void	open_next_file ()
Protected Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject ()=default
void	register_node (ElementId name, NewNodePtr)
void	publish (google::protobuf::Message &&, CustomOrigin &&co={}, OpMonLevel l=to_level(EntryOpMonLevel::kDefault)) const noexcept

Private Types
using	raw_sender_ct = iomanager::SenderConcept<ReadoutType>

Private Attributes
std::atomic< bool > &	m_run_marker
std::string	m_name
bool	m_is_configured = false
double	m_rate_khz
std::vector< std::string >	m_file_names
std::vector< std::string >::const_iterator	m_file_iterator
uint32_t	m_input_buffer_size
std::string	m_compression_algorithm
daqdataformats::SourceID	m_sourceid
std::atomic< int >	m_packet_count { 0 }
std::atomic< int >	m_packet_count_tot { 0 }
std::chrono::milliseconds	m_raw_sender_timeout_ms
std::shared_ptr< raw_sender_ct >	m_raw_data_sender
bool	m_sender_is_set = false
std::unique_ptr< dunedaq::datahandlinglibs::RateLimiter >	m_rate_limiter
std::unique_ptr< dunedaq::datahandlinglibs::BufferedFileReader< ReadoutType > >	m_file_reader
utilities::ReusableThread	m_producer_thread

Additional Inherited Members
Public Types inherited from dunedaq::opmonlib::MonitorableObject
using	NodePtr = std::weak_ptr<MonitorableObject>
using	NewNodePtr = std::shared_ptr<MonitorableObject>
using	ElementId = std::string
Static Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
static bool	publishable_metric (OpMonLevel entry, OpMonLevel system) noexcept

Detailed Description

template<class ReadoutType>
class dunedaq::snbmodules::FileSourceModel< ReadoutType >

Definition at line 43 of file FileSourceModel.hpp.

Member Typedef Documentation

raw_sender_ct

template<class ReadoutType >

using dunedaq::snbmodules::FileSourceModel< ReadoutType >::raw_sender_ct = iomanager::SenderConcept<ReadoutType>

private

Definition at line 106 of file FileSourceModel.hpp.

Constructor & Destructor Documentation

FileSourceModel()

template<class ReadoutType >

dunedaq::snbmodules::FileSourceModel< ReadoutType >::FileSourceModel ( std::string name, std::atomic< bool > & run_marker, double rate_khz )

inlineexplicit

Definition at line 46 of file FileSourceModel.hpp.

    : m_run_marker(run_marker)
    , m_name(name)
    , m_rate_khz(rate_khz)
    , m_packet_count{ 0 }
    , m_raw_sender_timeout_ms(0)
    , m_raw_data_sender(nullptr)
    , m_producer_thread(0)
  {
  }

Member Function Documentation

conf()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::conf ( const confmodel::DetectorStream * stream_conf, const appmodel::SNBFileSourceParameters * file_params )

virtual

Implements dunedaq::snbmodules::FileSourceConcept.

Definition at line 29 of file FileSourceModel.hxx.

{
  if (m_is_configured) {
    TLOG_DEBUG(TLVL_WORK_STEPS) << "This emulator is already configured!";
  } else {
    // m_conf = args.get<module_conf_t>();
    // m_link_conf = link_conf.get<link_conf_t>();
    m_raw_sender_timeout_ms = std::chrono::milliseconds(1000);
 
    m_sourceid.id = link_conf->get_source_id();
    m_sourceid.subsystem = ReadoutType::subsystem;
 
    m_file_names = file_params->get_data_files();
    m_file_iterator = m_file_names.begin();
 
    m_input_buffer_size = file_params->get_input_buffer_size();
    m_compression_algorithm = file_params->get_file_compression_algorithm();
 
    open_next_file();
 
    m_is_configured = true;
  }
  // Configure thread:
  m_producer_thread.set_name("fileread", m_sourceid.id);
}

generate_opmon_data()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::generate_opmon_data ( )

overrideprotectedvirtual

Hook for customisable pubblication. The function can throw, exception will be caught by the monitoring thread

Reimplemented from dunedaq::opmonlib::MonitorableObject.

Definition at line 102 of file FileSourceModel.hxx.

{
  datahandlinglibs::opmon::DataSourceInfo info;
  info.set_sum_packets(m_packet_count_tot.load());
  info.set_num_packets(m_packet_count.exchange(0));
 
  this->publish(std::move(info));
}

is_configured()

template<class ReadoutType >

bool dunedaq::snbmodules::FileSourceModel< ReadoutType >::is_configured ( )

inlineoverridevirtual

Implements dunedaq::snbmodules::FileSourceConcept.

Definition at line 69 of file FileSourceModel.hpp.

{ return m_is_configured; }

open_next_file()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::open_next_file ( )

protected

Definition at line 58 of file FileSourceModel.hxx.

{
  if (m_file_iterator == m_file_names.end()) {
    if (m_file_reader) {
      m_file_reader->close();
    }
    m_file_reader.reset();
    m_is_configured = false;
    return;
  }
  try {
    m_file_reader = std::make_unique<datahandlinglibs::BufferedFileReader<ReadoutType>>(
      *m_file_iterator, m_input_buffer_size, m_compression_algorithm);
    ++m_file_iterator;
  } catch (const ers::Issue& ex) {
    ers::fatal(ex);
    throw ConfigurationError(ERS_HERE, m_sourceid, "", ex);
  }
}

run_produce()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::run_produce ( )

protected

Definition at line 113 of file FileSourceModel.hxx.

{
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Data generation thread " << m_sourceid.to_string() << " started";
 
  // pthread_setname_np(pthread_self(), get_name().c_str());
 
  ReadoutType elem;
  while (m_run_marker.load()) {
    auto read_sts = m_file_reader->read(elem);
 
    if (!read_sts) {
      TLOG_DEBUG(TLVL_BOOKKEEPING) << "Failed to read, moving to next file";
      open_next_file();
      if (!m_is_configured) {
        break;
      }
      continue;
    }
 
    TLOG_DEBUG(TLVL_BOOKKEEPING) << "Read element with timestamp " << elem.get_timestamp() << " from file";
 
    // send it
    bool send_successful = false;
    while (!send_successful && m_run_marker.load()) {
      ReadoutType elem_copy(elem);
      send_successful = m_raw_data_sender->try_send(std::move(elem_copy), m_raw_sender_timeout_ms);
    }
 
    // Count packet and limit rate if needed.
    ++m_packet_count;
    ++m_packet_count_tot;
 
    m_rate_limiter->limit();
  }
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Data generation thread " << m_sourceid.to_string() << " finished";
}

scrap()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::scrap ( const appfwk::DAQModule::CommandData_t & )

inlinevirtual

Implements dunedaq::snbmodules::FileSourceConcept.

Definition at line 61 of file FileSourceModel.hpp.

  {
    if (m_file_reader != nullptr) {
      m_file_reader->close();
    }
    m_file_reader.reset();
    m_is_configured = false;
  }

set_sender()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::set_sender ( const std::string & conn_name )

virtual

Implements dunedaq::snbmodules::FileSourceConcept.

Definition at line 17 of file FileSourceModel.hxx.

{
  if (!m_sender_is_set) {
    m_raw_data_sender = get_iom_sender<ReadoutType>(conn_name);
    m_sender_is_set = true;
  } else {
    // ers::error();
  }
}

start()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::start ( const appfwk::DAQModule::CommandData_t & )

virtual

Implements dunedaq::snbmodules::FileSourceConcept.

Definition at line 80 of file FileSourceModel.hxx.

{
  m_packet_count_tot = 0;
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Starting threads...";
  // FIXME: don't know where to take the slowdown from... m_rate_limiter = std::make_unique<RateLimiter>(m_rate_khz /
  // m_link_conf.slowdown);
  m_rate_limiter = std::make_unique<datahandlinglibs::RateLimiter>(m_rate_khz);
  // m_stats_thread.set_work(&FileSourceModel<ReadoutType>::run_stats, this);
  m_producer_thread.set_work(&FileSourceModel<ReadoutType>::run_produce, this);
}

stop()

template<class ReadoutType >

void dunedaq::snbmodules::FileSourceModel< ReadoutType >::stop ( const appfwk::DAQModule::CommandData_t & )

virtual

Implements dunedaq::snbmodules::FileSourceConcept.

Definition at line 93 of file FileSourceModel.hxx.

{
  while (!m_producer_thread.get_readiness()) {
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
  }
}

Member Data Documentation

m_compression_algorithm

template<class ReadoutType >

std::string dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_compression_algorithm

private

Definition at line 94 of file FileSourceModel.hpp.

m_file_iterator

template<class ReadoutType >

std::vector<std::string>::const_iterator dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_file_iterator

private

Definition at line 92 of file FileSourceModel.hpp.

m_file_names

template<class ReadoutType >

std::vector<std::string> dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_file_names

private

Definition at line 91 of file FileSourceModel.hpp.

m_file_reader

template<class ReadoutType >

std::unique_ptr<dunedaq::datahandlinglibs::BufferedFileReader<ReadoutType> > dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_file_reader

private

Definition at line 116 of file FileSourceModel.hpp.

m_input_buffer_size

template<class ReadoutType >

uint32_t dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_input_buffer_size

private

Definition at line 93 of file FileSourceModel.hpp.

m_is_configured

template<class ReadoutType >

bool dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_is_configured = false

private

Definition at line 88 of file FileSourceModel.hpp.

m_name

template<class ReadoutType >

std::string dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_name

private

Definition at line 87 of file FileSourceModel.hpp.

m_packet_count

template<class ReadoutType >

std::atomic<int> dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_packet_count { 0 }

private

Definition at line 99 of file FileSourceModel.hpp.

{ 0 };

m_packet_count_tot

template<class ReadoutType >

std::atomic<int> dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_packet_count_tot { 0 }

private

Definition at line 100 of file FileSourceModel.hpp.

{ 0 };

m_producer_thread

template<class ReadoutType >

utilities::ReusableThread dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_producer_thread

private

Definition at line 118 of file FileSourceModel.hpp.

m_rate_khz

template<class ReadoutType >

double dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_rate_khz

private

Definition at line 89 of file FileSourceModel.hpp.

m_rate_limiter

template<class ReadoutType >

std::unique_ptr<dunedaq::datahandlinglibs::RateLimiter> dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_rate_limiter

private

Definition at line 115 of file FileSourceModel.hpp.

m_raw_data_sender

template<class ReadoutType >

std::shared_ptr<raw_sender_ct> dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_raw_data_sender

private

Definition at line 107 of file FileSourceModel.hpp.

m_raw_sender_timeout_ms

template<class ReadoutType >

std::chrono::milliseconds dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_raw_sender_timeout_ms

private

Definition at line 105 of file FileSourceModel.hpp.

m_run_marker

template<class ReadoutType >

std::atomic<bool>& dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_run_marker

private

Definition at line 84 of file FileSourceModel.hpp.

m_sender_is_set

template<class ReadoutType >

bool dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_sender_is_set = false

private

Definition at line 109 of file FileSourceModel.hpp.

m_sourceid

template<class ReadoutType >

daqdataformats::SourceID dunedaq::snbmodules::FileSourceModel< ReadoutType >::m_sourceid

private

Definition at line 96 of file FileSourceModel.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/snbmodules/include/snbmodules/readout/FileSourceModel.hpp
• /github/workspace/dunedaq/sourcecode/snbmodules/include/snbmodules/readout/detail/FileSourceModel.hxx