Line data    Source code

       1              : /**
       2              :  * @file SourceModel.hpp FELIX CR's ELink concept wrapper
       3              :  *
       4              :  * This is part of the DUNE DAQ , copyright 2020.
       5              :  * Licensing/copyright details are in the COPYING file that you should have
       6              :  * received with this code.
       7              :  */
       8              : #ifndef ASIOLIBS_SRC_SOURCEMODEL_HPP_
       9              : #define ASIOLIBS_SRC_SOURCEMODEL_HPP_
      10              : 
      11              : #include "SourceConcept.hpp"
      12              : 
      13              : 
      14              : #include "iomanager/IOManager.hpp"
      15              : #include "iomanager/Sender.hpp"
      16              : #include "logging/Logging.hpp"
      17              : 
      18              : #include "asiolibs/opmon/SourceModel.pb.h"
      19              : 
      20              : // #include "datahandlinglibs/utils/ReusableThread.hpp"
      21              : #include "datahandlinglibs/DataMoveCallbackRegistry.hpp"
      22              : #include "datahandlinglibs/utils/BufferCopy.hpp" 
      23              : 
      24              : // #include <folly/ProducerConsumerQueue.h>
      25              : // #include <nlohmann/json.hpp>
      26              : 
      27              : #include <atomic>
      28              : #include <memory>
      29              : #include <mutex>
      30              : #include <string>
      31              : #include <utility>
      32              : 
      33              : namespace dunedaq::asiolibs {
      34              : 
      35              : template<class TargetPayloadType>
      36              : class SourceModel : public SourceConcept
      37              : {
      38              : public:
      39              :   using sink_t = iomanager::SenderConcept<TargetPayloadType>;
      40              :   using inherited = SourceConcept;
      41              :   using data_t = nlohmann::json;
      42              : 
      43              :   /**
      44              :    * @brief SourceModel Constructor
      45              :    * @param name Instance name for this SourceModel instance
      46              :    */
      47            0 :   SourceModel()
      48            0 :     : SourceConcept()
      49            0 :   {}
      50            0 :   ~SourceModel() {}
      51              : 
      52            0 :   void acquire_callback() override
      53              :   {
      54            0 :       if (m_callback_is_acquired) {
      55            0 :         TLOG_DEBUG(5) << "SourceModel callback is already acquired!";
      56              :       } else {
      57              :         // Getting DataMoveCBRegistry
      58            0 :         auto dmcbr = datahandlinglibs::DataMoveCallbackRegistry::get();
      59            0 :         m_sink_callback = dmcbr->get_callback<TargetPayloadType>(inherited::m_sink_conf);
      60            0 :         m_callback_is_acquired = true;
      61            0 :       }
      62            0 :   }
      63              : 
      64              :   // Process an incoming raw byte buffer and extract complete frames of type TargetPayloadType.
      65            0 :   void handle_daq_frame(char* buffer) override
      66              :   {
      67              :     // Materialize a real TargetPayloadType object by copying bytes from the buffer.
      68              :     // This is defined behavior, alignment-safe, and fast, without pointer vodoo
      69              :     // Previously reinterpret_cast to TargetPayloadType* introduced alignment traps 
      70              :     // “pretend there’s a constructed object there” UB. 
      71              :     TargetPayloadType frame;
      72            0 :     std::memcpy(&frame, buffer, m_expected_frame_size);
      73              : 
      74              :      // Pass by value (moved); no references into 'buffer', so no UAF.
      75            0 :     (*m_sink_callback)(std::move(frame));
      76            0 :   }
      77              : 
      78            0 :   std::size_t get_expected_frame_size() const override {
      79            0 :     return m_expected_frame_size;
      80              :   }
      81              :     
      82            0 :   void generate_opmon_data() override {
      83              : 
      84            0 :     opmon::SourceInfo info;
      85            0 :     info.set_leftover_bytes_encountered( m_leftover_bytes_encountered.exchange(0) );
      86              : 
      87            0 :     publish( std::move(info) );
      88            0 :   }
      89              :   
      90              : private:
      91              :   // Constants
      92              :   const std::size_t m_expected_frame_size = sizeof(TargetPayloadType);
      93              : 
      94              :   // Callback internals
      95              :   bool m_callback_is_acquired{ false };
      96              :   using sink_cb_t = std::shared_ptr<std::function<void(TargetPayloadType&&)>>;
      97              :   sink_cb_t m_sink_callback;
      98              : 
      99              :   // Stats  
     100              :   std::atomic<uint64_t> m_leftover_bytes_encountered{0}; // NOLINT(build/unsigned)
     101              : 
     102              : };
     103              : 
     104              : } // namespace dunedaq::asiolibs
     105              : 
     106              : #endif // ASIOLIBS_SRC_SOURCEMODEL_HPP_