Line data Source code
1 : /**
2 : * @file TriggerInhibitAgent.cpp TriggerInhibitAgent class implementation
3 : *
4 : * This is part of the DUNE DAQ Software Suite, copyright 2020.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 :
9 : #include "dfmodules/TriggerInhibitAgent.hpp"
10 : #include "dfmodules/CommonIssues.hpp"
11 :
12 : #include "logging/Logging.hpp"
13 :
14 : #include <memory>
15 : #include <string>
16 : #include <utility>
17 :
18 : /**
19 : * @brief Name used by TRACE TLOG calls from this source file
20 : */
21 : #define TRACE_NAME "TriggerInhibitAgent" // NOLINT
22 : enum
23 : {
24 : TLVL_ENTER_EXIT_METHODS = 5,
25 : TLVL_WORK_STEPS = 10
26 : };
27 :
28 : namespace dunedaq {
29 : namespace dfmodules {
30 :
31 0 : TriggerInhibitAgent::TriggerInhibitAgent(const std::string& parent_name,
32 : std::shared_ptr<trigdecreceiver_t> our_input,
33 0 : std::shared_ptr<triginhsender_t> our_output)
34 0 : : NamedObject(parent_name + "::TriggerInhibitAgent")
35 0 : , m_thread(std::bind(&TriggerInhibitAgent::do_work, this, std::placeholders::_1))
36 0 : , m_queue_timeout(100)
37 0 : , m_threshold_for_inhibit(1)
38 0 : , m_trigger_decision_receiver(our_input)
39 0 : , m_trigger_inhibit_sender(our_output)
40 0 : , m_trigger_number_at_start_of_processing_chain(0)
41 0 : , m_trigger_number_at_end_of_processing_chain(0)
42 : {
43 0 : }
44 :
45 : void
46 0 : TriggerInhibitAgent::start_checking()
47 : {
48 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering start_checking() method";
49 0 : m_thread.start_working_thread();
50 0 : TLOG() << get_name() << " successfully started";
51 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting start_checking() method";
52 0 : }
53 :
54 : void
55 0 : TriggerInhibitAgent::stop_checking()
56 : {
57 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering stop_checking() method";
58 0 : m_thread.stop_working_thread();
59 0 : TLOG() << get_name() << " successfully stopped";
60 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting stop_checking() method";
61 0 : }
62 :
63 : void
64 0 : TriggerInhibitAgent::do_work(std::atomic<bool>& running_flag)
65 : {
66 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_work() method";
67 :
68 : // configuration (hard-coded, for now; will be input from calling code later)
69 0 : int fake_busy_interval_sec = 0;
70 0 : std::chrono::seconds chrono_fake_busy_interval(fake_busy_interval_sec);
71 0 : int fake_busy_duration_sec = 0;
72 0 : std::chrono::seconds chrono_fake_busy_duration(fake_busy_duration_sec);
73 0 : int min_interval_between_inhibit_messages_msec = 0;
74 0 : std::chrono::milliseconds chrono_min_interval_between_inhibit_messages(min_interval_between_inhibit_messages_msec);
75 :
76 : // initialization
77 0 : enum LocalState
78 : {
79 : no_update,
80 : free_state,
81 : busy_state
82 : };
83 0 : std::chrono::steady_clock::time_point current_time = std::chrono::steady_clock::now();
84 : // std::chrono::steady_clock::time_point start_time_of_latest_fake_busy = current_time - chrono_fake_busy_duration;
85 0 : std::chrono::steady_clock::time_point last_sent_time = current_time;
86 0 : LocalState requested_state = no_update;
87 0 : LocalState current_state = free_state;
88 0 : int32_t received_message_count = 0;
89 0 : int32_t sent_message_count = 0;
90 :
91 : // work loop
92 0 : while (running_flag.load()) {
93 :
94 : // check if a TriggerDecision message has arrived, and save the trigger
95 : // number contained within it, if one has arrived
96 0 : try {
97 0 : dfmessages::TriggerDecision trig_dec = m_trigger_decision_receiver->receive(m_queue_timeout);
98 0 : ++received_message_count;
99 0 : TLOG_DEBUG(TLVL_WORK_STEPS) << get_name() << ": Popped the TriggerDecision for trigger number "
100 0 : << trig_dec.trigger_number << " off the input queue";
101 0 : m_trigger_number_at_start_of_processing_chain.store(trig_dec.trigger_number);
102 0 : } catch (const iomanager::TimeoutExpired& excpt) {
103 : // it is perfectly reasonable that there will be no data in the queue some
104 : // fraction of the times that we check, so we just continue on and try again later
105 0 : }
106 :
107 : // to-do: add some logic to fake inhibits
108 :
109 : // check if A) we are supposed to be checking the trigger_number difference, and
110 : // B) if so, whether an Inhibit should be asserted or cleared
111 0 : uint32_t threshold = m_threshold_for_inhibit.load(); // NOLINT
112 0 : if (threshold > 0) {
113 0 : daqdataformats::trigger_number_t temp_trig_num_at_start = m_trigger_number_at_start_of_processing_chain.load();
114 0 : daqdataformats::trigger_number_t temp_trig_num_at_end = m_trigger_number_at_end_of_processing_chain.load();
115 0 : if (temp_trig_num_at_start >= temp_trig_num_at_end &&
116 0 : (temp_trig_num_at_start - temp_trig_num_at_end) >= threshold) {
117 0 : if (current_state == free_state) {
118 0 : requested_state = busy_state;
119 : }
120 : } else {
121 0 : if (current_state == busy_state) {
122 0 : requested_state = free_state;
123 : }
124 : }
125 : }
126 :
127 : // to-do: add some logic to periodically send a message even if nothing has changed
128 :
129 : // send an Inhibit messages, if needed (either Busy or Free state)
130 0 : if (requested_state != no_update && requested_state != current_state) {
131 0 : if ((std::chrono::steady_clock::now() - last_sent_time) >= chrono_min_interval_between_inhibit_messages) {
132 0 : dfmessages::TriggerInhibit inhibit_message;
133 0 : if (requested_state == busy_state) {
134 0 : inhibit_message.busy = true;
135 : } else {
136 : inhibit_message.busy = false;
137 : }
138 :
139 0 : TLOG_DEBUG(TLVL_WORK_STEPS) << get_name() << ": Pushing a TriggerInhibit message with busy state set to "
140 0 : << inhibit_message.busy << " onto the output queue";
141 0 : try {
142 0 : m_trigger_inhibit_sender->send(std::move(inhibit_message), m_queue_timeout);
143 0 : ++sent_message_count;
144 : #if 0
145 : // temporary logging
146 : std::ostringstream oss_sent;
147 : oss_sent << ": Successfully pushed a TriggerInhibit message with busy state set to " << inhibit_message.busy
148 : << " onto the output queue";
149 : TLOG() << ProgressUpdate(ERS_HERE, get_name(), oss_sent.str());
150 : #endif
151 : // if we successfully pushed the message to the Sink, then we assume that the
152 : // receiver will get it, and we update our internal state accordingly
153 0 : current_state = requested_state;
154 0 : requested_state = no_update;
155 0 : last_sent_time = std::chrono::steady_clock::now();
156 0 : } catch (const iomanager::TimeoutExpired& excpt) {
157 : // It is not ideal if we fail to send the inhibit message out, but rather than
158 : // retrying some unknown number of times, we simply output a TRACE message and
159 : // go on. This has the benefit of being responsive with pulling TriggerDecision
160 : // messages off the input queue, and maybe our Busy/Free state will have changed
161 : // by the time that the receiver is ready to receive more messages.
162 0 : TLOG_DEBUG(TLVL_WORK_STEPS) << get_name()
163 0 : << ": TIMEOUT pushing a TriggerInhibit message onto the output queue";
164 0 : }
165 : }
166 : }
167 : }
168 :
169 0 : std::ostringstream oss_summ;
170 0 : oss_summ << ": Exiting the do_work() method, received " << received_message_count
171 0 : << " TriggerDecision messages and sent " << sent_message_count
172 0 : << " TriggerInhibit messages of all types (both Busy and Free).";
173 0 : TLOG() << ProgressUpdate(ERS_HERE, get_name(), oss_summ.str());
174 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_work() method";
175 0 : }
176 :
177 : } // namespace dfmodules
178 : } // namespace dunedaq
|
