Line data    Source code

       1              : /**
       2              :  * @file FragmentAggregatorModule.hpp Module to dispatch data requests within an application, aggregate and send
       3              :  * fragments using the IOMManager
       4              :  *
       5              :  * This is part of the DUNE DAQ , copyright 2020.
       6              :  * Licensing/copyright details are in the COPYING file that you should have
       7              :  * received with this code.
       8              :  */
       9              : 
      10              : #ifndef DFMODULES_PLUGINS_FRAGMENTAGGREGATOR_HPP_
      11              : #define DFMODULES_PLUGINS_FRAGMENTAGGREGATOR_HPP_
      12              : 
      13              : #include "daqdataformats/Fragment.hpp"
      14              : #include "daqdataformats/SourceID.hpp"
      15              : #include "dfmessages/DataRequest.hpp"
      16              : 
      17              : #include "appfwk/DAQModule.hpp"
      18              : #include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
      19              : 
      20              : #include "iomanager/Receiver.hpp"
      21              : #include "iomanager/Sender.hpp"
      22              : 
      23              : #include <atomic>
      24              : #include <map>
      25              : #include <mutex>
      26              : #include <string>
      27              : #include <tuple>
      28              : 
      29              : namespace dunedaq {
      30              : /**
      31              :  * @brief Unknown TRB
      32              :  */
      33            0 : ERS_DECLARE_ISSUE(dfmodules,                  ///< Namespace
      34              :                   UnknownFragmentDestination, ///< Issue class name
      35              :                   "Could not find a valid destination for sending Fragment with trigger number: "
      36              :                     << trg_num << " sequence number: " << seq_num << " from DLH " << src, ///< Message
      37              :                   ((uint64_t)trg_num)                                                     ///< Message parameters
      38              :                   ((uint16_t)seq_num)                                                     ///< Message parameters
      39              :                   ((daqdataformats::SourceID)src)                                         ///< Message parameters
      40              : )
      41              : 
      42            0 : ERS_DECLARE_ISSUE(dfmodules,         ///< Namespace
      43              :                   AbandonedFragment, ///< Issue class name
      44              :                   "Fragment from " << source << " for trigger " << trigger << '-' << sequence << " of run " << run
      45              :                                    << " was dropped",
      46              :                   ((daqdataformats::run_number_t)run)((daqdataformats::trigger_number_t)trigger)(
      47              :                     (daqdataformats::sequence_number_t)sequence)((daqdataformats::SourceID)source))
      48              : 
      49              : namespace dfmodules {
      50              : 
      51              : class FragmentAggregatorModule : public dunedaq::appfwk::DAQModule
      52              : {
      53              : public:
      54              :   explicit FragmentAggregatorModule(const std::string& name);
      55              : 
      56              :   FragmentAggregatorModule(const FragmentAggregatorModule&) = delete;
      57              :   FragmentAggregatorModule& operator=(const FragmentAggregatorModule&) = delete;
      58              :   FragmentAggregatorModule(FragmentAggregatorModule&&) = delete;
      59              :   FragmentAggregatorModule& operator=(FragmentAggregatorModule&&) = delete;
      60              : 
      61              :   void init(std::shared_ptr<appfwk::ConfigurationManager> mcfg) override;
      62              :   void generate_opmon_data() override;
      63              : 
      64              : private:
      65              :   // Commands
      66              :   void do_start(const CommandData_t& obj);
      67              :   void do_stop(const CommandData_t& obj);
      68              : 
      69              :   void process_data_request(dfmessages::DataRequest&);
      70              :   void process_fragment(std::unique_ptr<daqdataformats::Fragment>&);
      71              : 
      72              :   // Input and Output Connection names
      73              :   std::string m_data_req_input;
      74              :   std::string m_fragment_input;
      75              :   std::map<int, std::string> m_producer_conn_ids;
      76              :   std::vector<std::string> m_trb_conn_ids;
      77              : 
      78              :   std::chrono::milliseconds m_fragment_send_timeout;
      79              : 
      80              :   // Opmon
      81              :   uint64_t get_current_time_us();
      82              :   uint64_t m_timestamp_before_dr;
      83              :   uint64_t m_timestamp_before_frag;
      84              :   using metric_counter_type = uint64_t;
      85              :   std::atomic<metric_counter_type> m_data_requests_received{ 0 };
      86              :   std::atomic<metric_counter_type> m_data_requests_processed{ 0 };
      87              :   std::atomic<metric_counter_type> m_data_requests_failed{ 0 };
      88              :   std::atomic<metric_counter_type> m_fragments_received{ 0 };
      89              :   std::atomic<metric_counter_type> m_fragments_processed{ 0 };
      90              :   std::atomic<metric_counter_type> m_fragments_failed{ 0 };
      91              :   std::atomic<metric_counter_type> m_fragments_empty{ 0 };
      92              :   std::atomic<metric_counter_type> m_fragments_incomplete{ 0 };
      93              :   std::atomic<metric_counter_type> m_fragments_invalid{ 0 };
      94              :   std::atomic<metric_counter_type> m_fragments_time_average_us{ 0 };
      95              :   std::atomic<metric_counter_type> m_fragments_time_min_us{ 0 };
      96              :   std::atomic<metric_counter_type> m_fragments_time_max_us{ 0 };
      97              :   std::atomic<metric_counter_type> m_data_requests_time_average_us{ 0 };
      98              :   std::atomic<metric_counter_type> m_data_requests_time_min_us{ 0 };
      99              :   std::atomic<metric_counter_type> m_data_requests_time_max_us{ 0 };
     100              : 
     101              :   // TRB tracking
     102              :   std::map<std::tuple<dfmessages::trigger_number_t, dfmessages::sequence_number_t, daqdataformats::SourceID>,
     103              :            std::string>
     104              :     m_data_req_map;
     105              :   std::mutex m_mutex;
     106              : };
     107              : } // namespace dfmodules
     108              : } // namespace dunedaq
     109              : 
     110              : #endif // DFMODULES_PLUGINS_FRAGMENTAGGREGATOR_HPP_