Line data    Source code

       1              : /**
       2              :  * @file TPStreamWriterModule.hpp
       3              :  *
       4              :  * TPStreamWriterModule is a DAQModule that provides sample code for writing TPSets to disk.
       5              :  *
       6              :  * This is part of the DUNE DAQ Software Suite, copyright 2020.
       7              :  * Licensing/copyright details are in the COPYING file that you should have
       8              :  * received with this code.
       9              :  */
      10              : 
      11              : #ifndef DFMODULES_PLUGINS_TPSTREAMWRITER_HPP_
      12              : #define DFMODULES_PLUGINS_TPSTREAMWRITER_HPP_
      13              : 
      14              : #include "dfmodules/DataStore.hpp"
      15              : 
      16              : #include "appfwk/DAQModule.hpp"
      17              : #include "appmodel/TPStreamWriterConf.hpp"
      18              : #include "daqdataformats/TimeSlice.hpp"
      19              : #include "iomanager/Receiver.hpp"
      20              : #include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
      21              : #include "trigger/TPSet.hpp"
      22              : #include "utilities/WorkerThread.hpp"
      23              : 
      24              : #include <memory>
      25              : #include <string>
      26              : 
      27              : namespace dunedaq {
      28              : namespace dfmodules {
      29              : 
      30              : /**
      31              :  * @brief TPStreamWriterModule receives TPSets from a queue and prints them out
      32              :  */
      33              : class TPStreamWriterModule : public dunedaq::appfwk::DAQModule
      34              : {
      35              : public:
      36              :   /**
      37              :    * @brief TPStreamWriterModule Constructor
      38              :    * @param name Instance name for this TPStreamWriterModule instance
      39              :    */
      40              :   explicit TPStreamWriterModule(const std::string& name);
      41              : 
      42              :   TPStreamWriterModule(const TPStreamWriterModule&) = delete; ///< TPStreamWriterModule is not copy-constructible
      43              :   TPStreamWriterModule& operator=(const TPStreamWriterModule&) =
      44              :     delete;                                                         ///< TPStreamWriterModule is not copy-assignable
      45              :   TPStreamWriterModule(TPStreamWriterModule&&) = delete;            ///< TPStreamWriterModule is not move-constructible
      46              :   TPStreamWriterModule& operator=(TPStreamWriterModule&&) = delete; ///< TPStreamWriterModule is not move-assignable
      47              : 
      48              :   void init(std::shared_ptr<appfwk::ConfigurationManager> mcfg) override;
      49              :   void generate_opmon_data() override;
      50              : 
      51              : private:
      52              :   // Commands
      53              :   void do_conf(const CommandData_t&);
      54              :   void do_start(const CommandData_t&);
      55              :   void do_stop(const CommandData_t&);
      56              :   void do_scrap(const CommandData_t&);
      57              : 
      58              :   // Threading
      59              :   dunedaq::utilities::WorkerThread m_thread;
      60              :   void do_work(std::atomic<bool>&);
      61              : 
      62              :   // Configuration
      63              : 
      64              :   std::shared_ptr<appfwk::ConfigurationManager> m_module_configuration;
      65              :   const appmodel::TPStreamWriterConf* m_tp_writer_conf;
      66              :   std::chrono::milliseconds m_queue_timeout;
      67              :   size_t m_accumulation_interval_ticks;
      68              :   std::chrono::steady_clock::duration m_accumulation_inactivity_time_before_write;
      69              :   daqdataformats::run_number_t m_run_number;
      70              :   uint32_t m_source_id; // NOLINT(build/unsigned)
      71              :   bool m_warn_user_when_tardy_tps_are_discarded;
      72              :   double m_accumulation_interval_seconds;
      73              :   std::string m_writer_identifier;
      74              :   bool m_data_storage_is_enabled;
      75              : 
      76              :   // Queue sources and sinks
      77              :   using source_t = iomanager::ReceiverConcept<trigger::TPSet>;
      78              :   std::shared_ptr<source_t> m_tpset_source;
      79              : 
      80              :   // Worker(s)
      81              :   std::shared_ptr<DataStore> m_data_writer;
      82              : 
      83              :   // Metrics
      84              :   std::atomic<uint64_t> m_heartbeat_tpsets = { 0 };            // NOLINT(build/unsigned)
      85              :   std::atomic<uint64_t> m_tpsets_with_tps = { 0 };             // NOLINT(build/unsigned)
      86              :   std::atomic<uint64_t> m_tps_received = { 0 };                // NOLINT(build/unsigned)
      87              :   std::atomic<uint64_t> m_tps_written = { 0 };                 // NOLINT(build/unsigned)
      88              :   std::atomic<uint64_t> m_tps_discarded = { 0 };               // NOLINT(build/unsigned)
      89              :   std::atomic<uint64_t> m_timeslices_written = { 0 };          // NOLINT(build/unsigned)
      90              :   std::atomic<uint64_t> m_bytes_output = { 0 };                // NOLINT(build/unsigned)
      91              :   std::atomic<double> m_tardy_timeslice_max_seconds = { 0.0 }; // NOLINT(build/unsigned)
      92              :   std::atomic<uint64_t> m_total_tps_received = { 0 };          // NOLINT(build/unsigned)
      93              :   std::atomic<uint64_t> m_total_tps_written = { 0 };           // NOLINT(build/unsigned)
      94              :   std::atomic<uint64_t> m_total_tps_discarded = { 0 };         // NOLINT(build/unsigned)
      95              : };
      96              : } // namespace dfmodules
      97              : 
      98            0 : ERS_DECLARE_ISSUE_BASE(dfmodules,
      99              :                        InvalidDataWriterModule,
     100              :                        appfwk::GeneralDAQModuleIssue,
     101              :                        "A valid dataWriter instance is not available so it will not be possible to write data. A "
     102              :                        "likely cause for this is a skipped or missed Configure transition.",
     103              :                        ((std::string)name),
     104              :                        ERS_EMPTY)
     105              : 
     106            0 : ERS_DECLARE_ISSUE_BASE(dfmodules,
     107              :                        DataWritingProblem,
     108              :                        appfwk::GeneralDAQModuleIssue,
     109              :                        "A problem was encountered when writing TimeSlice number " << trnum << " in run " << runnum,
     110              :                        ((std::string)name),
     111              :                        ((size_t)trnum)((size_t)runnum))
     112              : 
     113            0 : ERS_DECLARE_ISSUE_BASE(dfmodules,
     114              :                        TardyTPsDiscarded,
     115              :                        appfwk::GeneralDAQModuleIssue,
     116              :                        "Tardy TPs from SourceIDs [" << sid_list << "] were discarded from TimeSlice number " << trnum
     117              :                                                     << " (~" << sec_too_late << " sec too late)",
     118              :                        ((std::string)name),
     119              :                        ((std::string)sid_list)((size_t)trnum)((float)sec_too_late))
     120              : 
     121              : } // namespace dunedaq
     122              : 
     123              : #endif // DFMODULES_PLUGINS_TPSTREAMWRITER_HPP_