SourceEmulatorModel.hxx

Go to the documentation of this file.

// Declarations for SourceEmulatorModel
 
using dunedaq::datahandlinglibs::logging::TLVL_TAKE_NOTE;
using dunedaq::datahandlinglibs::logging::TLVL_WORK_STEPS;
 
namespace dunedaq {
namespace datahandlinglibs {
 
void
SourceEmulatorPatternGenerator::generate(int source_id, int size)
{
  // TLOG() << "Generate random ADC patterns" ;
  std::srand(source_id * 12345);
  m_channel.reserve(size);
  for (int i = 0; i < size; i++) {
    int random_ch = std::rand() % 64;
    m_channel.push_back(random_ch);
  }
}
 
template<class ReadoutType>
void
SourceEmulatorModel<ReadoutType>::acquire_callback()
{
  if (!m_sender_is_set) {
    m_raw_data_callback =
      datahandlinglibs::DataMoveCallbackRegistry::get()->get_callback<ReadoutType>(SourceEmulatorConcept::m_sink_conf);
    m_sender_is_set = true;
  } else {
    // ers::error();
  }
}
 
template<class ReadoutType>
void
SourceEmulatorModel<ReadoutType>::conf(const confmodel::DetectorStream* link_conf,
                                       const appmodel::StreamEmulationParameters* emu_params)
{
  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>();
 
    std::mt19937 mt(rand()); // NOLINT(runtime/threadsafe_fn)
    std::uniform_real_distribution<double> dis(0.0, 1.0);
 
    m_sourceid.id = link_conf->get_source_id();
    m_sourceid.subsystem = ReadoutType::subsystem;
 
    m_crateid = link_conf->get_geo_id()->get_crate_id();
    m_slotid = link_conf->get_geo_id()->get_slot_id();
    m_linkid = link_conf->get_geo_id()->get_stream_id();
 
    m_t0_now = emu_params->get_set_t0();
    m_file_source = std::make_unique<FileSourceBuffer>(emu_params->get_input_file_size_limit(), sizeof(ReadoutType));
    try {
      m_file_source->read(emu_params->get_data_file_name());
    } catch (const ers::Issue& ex) {
      ers::fatal(ex);
      throw ConfigurationError(ERS_HERE, m_sourceid, "", ex);
    }
    m_dropouts_length = emu_params->get_random_population_size();
    if (m_dropout_rate == 0.0) {
      m_dropouts = std::vector<bool>(1);
    } else {
      m_dropouts = std::vector<bool>(m_dropouts_length);
    }
    for (size_t i = 0; i < m_dropouts.size(); ++i) {
      m_dropouts[i] = dis(mt) >= m_dropout_rate;
    }
 
    m_frame_errors_length = emu_params->get_random_population_size();
    m_frame_error_rate = emu_params->get_frame_error_rate_hz();
    m_error_bit_generator = ErrorBitGenerator(m_frame_error_rate);
    m_error_bit_generator.generate();
 
    // Generate random ADC pattern
    m_generate_periodic_adc_pattern = emu_params->get_generate_periodic_adc_pattern();
    auto vec_size = emu_params->get_random_population_size();
    if (m_generate_periodic_adc_pattern) {
      TLOG() << "Generated pattern.";
      m_pattern_generator.generate(m_sourceid.id, vec_size);
 
      if (emu_params->get_TP_rate_per_channel() != 0) {
        TLOG() << "TP rate per channel multiplier (base of 100 Hz/ch): " << emu_params->get_TP_rate_per_channel();
        // Define time to wait when adding an ADC above threshold
        // Adding a hit every 9768 gives a total Sent TP rate of approx 100 Hz/wire with WIBEth
        m_time_to_wait = m_time_to_wait / emu_params->get_TP_rate_per_channel();
      }
    }
 
    m_is_configured = true;
  }
  // Configure thread:
  m_producer_thread.set_name("fakeprod", m_sourceid.id);
}
 
template<class ReadoutType>
void
SourceEmulatorModel<ReadoutType>::start(const appfwk::DAQModule::CommandData_t& /*args*/)
{
  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<RateLimiter>(m_rate_khz);
  // m_stats_thread.set_work(&SourceEmulatorModel<ReadoutType>::run_stats, this);
  m_producer_thread.set_work(&SourceEmulatorModel<ReadoutType>::run_produce, this);
}
 
template<class ReadoutType>
void
SourceEmulatorModel<ReadoutType>::stop(const appfwk::DAQModule::CommandData_t& /*args*/)
{
  while (!m_producer_thread.get_readiness()) {
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
  }
}
 
template<class ReadoutType>
void
SourceEmulatorModel<ReadoutType>::generate_opmon_data()
{
  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));
}
 
template<class ReadoutType>
void
SourceEmulatorModel<ReadoutType>::run_produce()
{
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Data generation thread " << m_this_link_number << " started";
 
  // pthread_setname_np(pthread_self(), get_name().c_str());
 
  uint offset = 0; // NOLINT(build/unsigned)
  auto& source = m_file_source->get();
 
  uint num_elem = m_file_source->num_elements();
  if (num_elem == 0) {
    TLOG_DEBUG(TLVL_WORK_STEPS) << "No elements to read from buffer! Sleeping...";
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    num_elem = m_file_source->num_elements();
  }
 
  auto rptr = reinterpret_cast<ReadoutType*>(source.data()); // NOLINT
 
  // set the initial timestamp to a configured value, otherwise just use the timestamp from the header
  uint64_t ts_0 = rptr->get_timestamp(); // NOLINT(build/unsigned)
  if (m_t0_now) {
    auto time_now = std::chrono::system_clock::now().time_since_epoch();
    uint64_t current_time = // NOLINT (build/unsigned)
      std::chrono::duration_cast<std::chrono::microseconds>(time_now).count();
    // FIXME: where do I get the clockspeed from?
    // ts_0 = (m_conf.clock_speed_hz / 100000) * current_time;
    ts_0 = 625 * current_time / 10;
  }
  TLOG_DEBUG(TLVL_BOOKKEEPING) << "Using first timestamp: " << ts_0;
  uint64_t timestamp = ts_0; // NOLINT(build/unsigned)
  int dropout_index = 0;
  uint64_t number_pattern_hits_generated = 0;
  // 64 total channels, placing on slot 0 gives 64 available slots.
  const uint64_t max_tps_per_frame = 64;
 
  while (m_run_marker.load()) {
    // TLOG() << "Generating " << m_frames_per_tick << " for TS " << timestamp;
    for (uint16_t i = 0; i < m_frames_per_tick; i++) {
      // Which element to push to the buffer
      if (offset == num_elem || (offset + 1) * sizeof(ReadoutType) > source.size()) {
        offset = 0;
      }
 
      bool create_frame = m_dropouts[dropout_index]; // NOLINT(runtime/threadsafe_fn)
      dropout_index = (dropout_index + 1) % m_dropouts.size();
      if (create_frame) {
        ReadoutType payload;
        // Memcpy from file buffer to flat char array
        ::memcpy(static_cast<void*>(&payload),
                 static_cast<void*>(source.data() + offset * sizeof(ReadoutType)),
                 sizeof(ReadoutType));
 
        // Fake timestamp
        payload.fake_timestamps(timestamp, m_time_tick_diff);
 
        // Fake geoid
        payload.fake_geoid(m_crateid, m_slotid, m_linkid);
 
        // Introducing frame errors
        std::vector<uint16_t> frame_errs; // NOLINT(build/unsigned)
        for (size_t i = 0; i < rptr->get_num_frames(); ++i) {
          frame_errs.push_back(m_error_bit_generator.next());
        }
        payload.fake_frame_errors(&frame_errs);
 
        if (m_generate_periodic_adc_pattern) {
          uint64_t number_pattern_hits_expected = (timestamp - ts_0) / m_time_to_wait;
 
          // Calculate how many TPs to generate in this frame
          uint64_t tps_this_frame = 0;
          if (number_pattern_hits_expected > number_pattern_hits_generated) {
            tps_this_frame = number_pattern_hits_expected - number_pattern_hits_generated;
          }
 
          if (tps_this_frame > max_tps_per_frame) {
            tps_this_frame = max_tps_per_frame;
          }
 
          // Distribute TPs across channels via the pattern generator.
          for (uint64_t tp_idx = 0; tp_idx < tps_this_frame; ++tp_idx) {
            int channel = m_pattern_generator.get_channel_number();
            // The pattern generator draws channel in range 0-63, current
            // behaviour for frame type with 32 channels is silent dropping.
            try {
              payload.fake_adc_pattern(channel);
            } catch (const std::out_of_range&) {
            }
          }
 
          // Count the number of patterns attempted to inject. Prevents
          // expected - generated deficit from accumulating in case of
          // injection failure.
          number_pattern_hits_generated += tps_this_frame;
        }
 
        // send it
        try {
          (*m_raw_data_callback)(std::move(payload));
        } catch (ers::Issue& excpt) {
          ers::warning(CannotWriteToQueue(ERS_HERE, m_sourceid, "raw data input queue", excpt));
          // std::runtime_error("Queue timed out...");
        }
 
        // Count packet and limit rate if needed.
        ++offset;
        ++m_packet_count;
        ++m_packet_count_tot;
      }
    }
    timestamp += m_time_tick_diff * rptr->get_num_frames();
 
    m_rate_limiter->limit();
  }
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Data generation thread " << m_sourceid.to_string() << " finished";
}
 
} // namespace datahandlinglibs
} // namespace dunedaq