Line data    Source code

       1              : #include "trigger/TPRequestHandler.hpp"
       2              : #include "appmodel/DataHandlerConf.hpp"
       3              : #include "appmodel/RequestHandler.hpp"
       4              : 
       5              : #include "rcif/cmd/Nljs.hpp"
       6              : 
       7              : namespace dunedaq {
       8              : namespace trigger {
       9              : 
      10              : void
      11            0 : TPRequestHandler::conf(const appmodel::DataHandlerModule* conf)
      12              : {
      13              : 
      14            0 :   for (auto output : conf->get_outputs()) {
      15            0 :     if (output->get_data_type() == "TPSet") {
      16            0 :       try {
      17            0 :         m_tpset_sink = iomanager::IOManager::get()->get_sender<dunedaq::trigger::TPSet>(output->UID());
      18            0 :       } catch (const ers::Issue& excpt) {
      19            0 :         throw datahandlinglibs::ResourceQueueError(ERS_HERE, "tp queue", "DefaultRequestHandlerModel", excpt);
      20            0 :       }
      21              :     }
      22              :   }
      23            0 :   inherited2::conf(conf);
      24            0 : }
      25              : 
      26              : void
      27            0 : TPRequestHandler::scrap(const appfwk::DAQModule::CommandData_t& args)
      28              : {
      29            0 :   m_tpset_sink.reset();
      30            0 :   inherited2::scrap(args);
      31            0 : }
      32              : 
      33              : void
      34            0 : TPRequestHandler::start(const appfwk::DAQModule::CommandData_t& args)
      35              : {
      36              : 
      37            0 :   m_oldest_ts = 0;
      38            0 :   m_newest_ts = 0;
      39            0 :   m_start_win_ts = 0;
      40            0 :   m_end_win_ts = 0;
      41            0 :   m_first_cycle = true;
      42              : 
      43            0 :   inherited2::start(args);
      44            0 :   rcif::cmd::StartParams start_params = args.get<rcif::cmd::StartParams>();
      45            0 :   m_run_number = start_params.run;
      46            0 : }
      47              : 
      48              : void
      49            0 : TPRequestHandler::periodic_data_transmission()
      50              : {
      51              : 
      52            0 :   if (m_tpset_sink == nullptr)
      53              :     return;
      54              : 
      55            0 :   dunedaq::dfmessages::DataRequest dr;
      56              : 
      57            0 :   {
      58            0 :     std::unique_lock<std::mutex> lock(m_cv_mutex);
      59            0 :     m_cv.wait(lock, [&] { return !m_cleanup_requested; });
      60            0 :     m_requests_running++;
      61            0 :   }
      62            0 :   m_cv.notify_all();
      63            0 :   if (m_latency_buffer->occupancy() != 0) {
      64              :     // Prepare response
      65            0 :     RequestResult rres(ResultCode::kUnknown, dr);
      66            0 :     std::vector<std::pair<void*, size_t>> frag_pieces;
      67              : 
      68              :     // Get the newest TP
      69            0 :     SkipListAcc acc(inherited2::m_latency_buffer->get_skip_list());
      70            0 :     auto tail = acc.last();
      71            0 :     auto head = acc.first();
      72            0 :     m_newest_ts = (*tail).get_timestamp();
      73            0 :     m_oldest_ts = (*head).get_timestamp();
      74              : 
      75            0 :     if (m_first_cycle) {
      76            0 :       m_start_win_ts = m_oldest_ts;
      77            0 :       m_first_cycle = false;
      78              :     }
      79            0 :     if (m_newest_ts - m_start_win_ts > m_ts_set_sender_offset_ticks) {
      80            0 :       m_end_win_ts = m_newest_ts - m_ts_set_sender_offset_ticks;
      81            0 :       frag_pieces = get_fragment_pieces(m_start_win_ts, m_end_win_ts, rres);
      82            0 :       auto num_tps = frag_pieces.size();
      83            0 :       trigger::TPSet tpset;
      84            0 :       tpset.run_number = m_run_number;
      85            0 :       tpset.type = num_tps > 0 ? trigger::TPSet::Type::kPayload : trigger::TPSet::Type::kHeartbeat;
      86            0 :       tpset.origin = m_sourceid;
      87            0 :       tpset.start_time = m_start_win_ts;  // provisory timestamp, will be filled with first TP
      88            0 :       tpset.end_time = m_end_win_ts;      // provisory timestamp, will be filled with last TP
      89            0 :       tpset.seqno = m_next_tpset_seqno++; // NOLINT(runtime/increment_decrement)
      90              :                                           // reserve the space for efficiency
      91            0 :       if (num_tps > 0) {
      92            0 :         tpset.objects.reserve(frag_pieces.size());
      93            0 :         bool first_tp = true;
      94            0 :         for (auto f : frag_pieces) {
      95            0 :           trgdataformats::TriggerPrimitive tp = *(static_cast<trgdataformats::TriggerPrimitive*>(f.first));
      96              : 
      97            0 :           if (first_tp) {
      98            0 :             tpset.start_time = tp.time_start;
      99            0 :             first_tp = false;
     100              :           }
     101            0 :           tpset.end_time = tp.time_start;
     102            0 :           tpset.objects.emplace_back(std::move(tp));
     103              :         }
     104              :       }
     105            0 :       if (!m_tpset_sink->try_send(std::move(tpset), iomanager::Sender::s_no_block)) {
     106            0 :         ers::warning(DroppedTPSet(ERS_HERE, m_start_win_ts, m_end_win_ts));
     107            0 :         m_num_periodic_send_failed++;
     108              :       }
     109            0 :       m_num_periodic_sent++;
     110              : 
     111              :       // remember what we sent for the next loop
     112            0 :       m_start_win_ts = m_end_win_ts;
     113            0 :     }
     114            0 :   }
     115            0 :   {
     116            0 :     std::lock_guard<std::mutex> lock(m_cv_mutex);
     117            0 :     m_requests_running--;
     118            0 :   }
     119            0 :   m_cv.notify_all();
     120            0 :   return;
     121            0 : }
     122              : 
     123              : } // namespace fdreadoutlibs
     124              : } // namespace dunedaq