Line data Source code
1 : /**
2 : * @file ElinkModel.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 FLXLIBS_SRC_ELINKMODEL_HPP_
9 : #define FLXLIBS_SRC_ELINKMODEL_HPP_
10 :
11 : #include "ElinkConcept.hpp"
12 :
13 : #include "flxlibs/opmon/ElinkModel.pb.h"
14 :
15 : #include "packetformat/block_format.hpp"
16 :
17 : // #include "appfwk/DAQModuleHelper.hpp"
18 : #include "iomanager/IOManager.hpp"
19 : #include "iomanager/Sender.hpp"
20 : #include "logging/Logging.hpp"
21 : #include "utilities/ReusableThread.hpp"
22 :
23 : #include "datahandlinglibs/DataMoveCallbackRegistry.hpp"
24 :
25 : #include <folly/ProducerConsumerQueue.h>
26 : #include <nlohmann/json.hpp>
27 :
28 : #include <atomic>
29 : #include <memory>
30 : #include <mutex>
31 : #include <string>
32 :
33 : namespace dunedaq::flxlibs {
34 :
35 : template<class TargetPayloadType>
36 : class ElinkModel : public ElinkConcept
37 : {
38 : public:
39 : using err_sink_t = iomanager::SenderConcept<felix::packetformat::chunk>;
40 : using inherited = ElinkConcept;
41 : using data_t = nlohmann::json;
42 :
43 : /**
44 : * @brief ElinkModel Constructor
45 : * @param name Instance name for this ElinkModel instance
46 : */
47 0 : ElinkModel()
48 : : ElinkConcept()
49 0 : , m_run_marker{ false }
50 0 : , m_parser_thread(0)
51 : {
52 0 : }
53 0 : ~ElinkModel() {}
54 :
55 : std::shared_ptr<err_sink_t>& get_error_sink() { return m_error_sink_queue; }
56 :
57 0 : void init(const size_t block_queue_capacity)
58 : {
59 0 : m_block_addr_queue = std::make_unique<folly::ProducerConsumerQueue<uint64_t>>(block_queue_capacity); // NOLINT
60 0 : }
61 :
62 0 : void conf(size_t block_size, bool is_32b_trailers)
63 : {
64 0 : if (m_configured) {
65 0 : TLOG_DEBUG(5) << "ElinkModel is already configured!";
66 : } else {
67 0 : m_parser_thread.set_name(inherited::m_elink_source_tid, inherited::m_link_tag);
68 : // if (inconsistency)
69 : // ers::fatal(ElinkConfigurationInconsistency(ERS_HERE, m_num_links));
70 :
71 0 : m_parser->configure(block_size, is_32b_trailers); // unsigned bsize, bool trailer_is_32bit
72 0 : m_configured = true;
73 : }
74 0 : }
75 :
76 0 : void start()
77 : {
78 0 : m_t0 = std::chrono::high_resolution_clock::now();
79 0 : if (!m_run_marker.load()) {
80 0 : set_running(true);
81 0 : m_parser_thread.set_work(&ElinkModel::process_elink, this);
82 0 : TLOG() << "Started ElinkModel of link " << inherited::m_link_id << "...";
83 : } else {
84 0 : TLOG_DEBUG(5) << "ElinkModel of link " << inherited::m_link_id << " is already running!";
85 : }
86 0 : }
87 :
88 0 : void stop()
89 : {
90 0 : if (m_run_marker.load()) {
91 0 : set_running(false);
92 0 : while (!m_parser_thread.get_readiness()) {
93 0 : std::this_thread::sleep_for(std::chrono::milliseconds(10));
94 : }
95 0 : TLOG_DEBUG(5) << "Stopped ElinkModel of link " << m_link_id << "!";
96 : } else {
97 0 : TLOG_DEBUG(5) << "ElinkModel of link " << m_link_id << " is already stopped!";
98 : }
99 0 : }
100 :
101 0 : void set_running(bool should_run)
102 : {
103 0 : bool was_running = m_run_marker.exchange(should_run);
104 0 : TLOG_DEBUG(5) << "Active state was toggled from " << was_running << " to " << should_run;
105 0 : }
106 :
107 0 : bool queue_in_block_address(uint64_t block_addr) // NOLINT(build/unsigned)
108 : {
109 0 : if (m_block_addr_queue->write(block_addr)) { // ok write
110 : return true;
111 : } else { // failed write
112 : return false;
113 : }
114 : }
115 :
116 0 : void acquire_callback() override
117 : {
118 0 : if (m_callback_is_acquired) {
119 0 : TLOG_DEBUG(5) << "SourceModel callback is already acquired!";
120 : } else {
121 : // Getting DataMoveCBRegistry
122 0 : auto dmcbr = datahandlinglibs::DataMoveCallbackRegistry::get();
123 0 : m_sink_callback = dmcbr->get_callback<TargetPayloadType>(inherited::m_sink_conf);
124 0 : m_callback_is_acquired = true;
125 0 : }
126 0 : }
127 :
128 : // Callbacks
129 : bool m_callback_is_acquired{ false };
130 : using sink_cb_t = std::shared_ptr<std::function<void(TargetPayloadType&&)>>;
131 : sink_cb_t m_sink_callback;
132 :
133 : protected:
134 0 : void generate_opmon_data() override
135 : {
136 :
137 0 : opmon::CardReaderInfo info;
138 0 : auto now = std::chrono::high_resolution_clock::now();
139 0 : auto& stats = m_parser_impl.get_stats();
140 :
141 0 : double seconds = std::chrono::duration_cast<std::chrono::microseconds>(now - m_t0).count() / 1000000.;
142 :
143 0 : info.set_num_short_chunks_processed(stats.short_ctr.exchange(0));
144 0 : info.set_num_chunks_processed(stats.chunk_ctr.exchange(0));
145 0 : info.set_num_subchunks_processed(stats.subchunk_ctr.exchange(0));
146 0 : info.set_num_blocks_processed(stats.block_ctr.exchange(0));
147 :
148 0 : info.set_rate_blocks_processed(info.num_blocks_processed() / seconds / 1000.);
149 0 : info.set_rate_chunks_processed(info.num_chunks_processed() / seconds / 1000.);
150 :
151 0 : info.set_num_short_chunks_processed_with_error(stats.error_short_ctr.exchange(0));
152 0 : info.set_num_chunks_processed_with_error(stats.error_chunk_ctr.exchange(0));
153 0 : info.set_num_subchunks_processed_with_error(stats.error_subchunk_ctr.exchange(0));
154 0 : info.set_num_blocks_processed_with_error(stats.error_block_ctr.exchange(0));
155 0 : info.set_num_subchunk_crc_errors(stats.subchunk_crc_error_ctr.exchange(0));
156 0 : info.set_num_subchunk_trunc_errors(stats.subchunk_trunc_error_ctr.exchange(0));
157 0 : info.set_num_subchunk_errors(stats.subchunk_error_ctr.exchange(0));
158 :
159 0 : TLOG_DEBUG(2) << inherited::m_elink_str // Move to TLVL_TAKE_NOTE from readout
160 0 : << " Parser stats ->"
161 0 : << " Blocks: " << info.num_blocks_processed() << " Block rate: " << info.rate_blocks_processed()
162 0 : << " [kHz]"
163 0 : << " Chunks: " << info.num_chunks_processed() << " Chunk rate: " << info.rate_chunks_processed()
164 0 : << " [kHz]"
165 0 : << " Shorts: " << info.num_short_chunks_processed() << " Subchunks:" << info.num_subchunks_processed()
166 0 : << " Error Chunks: " << info.num_chunks_processed_with_error()
167 0 : << " Error Shorts: " << info.num_short_chunks_processed_with_error()
168 0 : << " Error Subchunks: " << info.num_subchunks_processed_with_error()
169 0 : << " Error Block: " << info.num_blocks_processed_with_error();
170 :
171 0 : m_t0 = now;
172 :
173 0 : publish(std::move(info),
174 0 : { { "card", std::to_string(m_card_id) },
175 0 : { "logical_unit", std::to_string(m_logical_unit) },
176 0 : { "link", std::to_string(m_link_id) },
177 0 : { "tag", std::to_string(m_link_tag) } });
178 0 : }
179 :
180 : private:
181 : // Types
182 : using UniqueBlockAddrQueue = std::unique_ptr<folly::ProducerConsumerQueue<uint64_t>>; // NOLINT(build/unsigned)
183 :
184 : // Internals
185 : std::atomic<bool> m_run_marker;
186 : bool m_configured{ false };
187 :
188 : // Sink
189 : bool m_sink_is_set{ false };
190 : std::shared_ptr<err_sink_t> m_error_sink_queue;
191 :
192 : // blocks to process
193 : UniqueBlockAddrQueue m_block_addr_queue;
194 :
195 : // Processor
196 : inline static const std::string m_parser_thread_name = "elinkp";
197 : utilities::ReusableThread m_parser_thread;
198 0 : void process_elink()
199 : {
200 0 : while (m_run_marker.load()) {
201 : uint64_t block_addr; // NOLINT
202 0 : if (m_block_addr_queue->read(block_addr)) { // read success
203 : const auto* block = const_cast<felix::packetformat::block*>(
204 0 : felix::packetformat::block_from_bytes(reinterpret_cast<const char*>(block_addr)) // NOLINT
205 : );
206 0 : m_parser->process(block);
207 : } else { // couldn't read from queue
208 0 : std::this_thread::sleep_for(std::chrono::milliseconds(10));
209 : }
210 : }
211 0 : }
212 : };
213 :
214 : } // namespace dunedaq::flxlibs
215 :
216 : #endif // FLXLIBS_SRC_ELINKMODEL_HPP_
|
