Line data Source code
1 : #include "artdaq/DAQdata/Globals.hh"
2 : #define TRACE_NAME (app_name + "_RequestBuffer").c_str() // include these 2 first -
3 :
4 : #include "artdaq/DAQrate/RequestBuffer.hh"
5 :
6 27 : artdaq::RequestBuffer::RequestBuffer()
7 :
8 27 : : requests_()
9 27 : , request_timing_()
10 27 : , next_requests_()
11 54 : , receiver_running_(false)
12 : {
13 27 : }
14 :
15 27 : artdaq::RequestBuffer::~RequestBuffer() {}
16 :
17 400072 : void artdaq::RequestBuffer::push(artdaq::Fragment::sequence_id_t seq, artdaq::Fragment::timestamp_t ts)
18 : {
19 400072 : std::lock_guard<std::mutex> tlk(request_mutex_);
20 400072 : if (requests_.count(seq) && requests_[seq] != ts)
21 : {
22 0 : TLOG(TLVL_ERROR) << "Received conflicting request for SeqID "
23 0 : << seq << "!"
24 0 : << " Old ts=" << requests_[seq]
25 0 : << ", new ts=" << ts << ". Keeping OLD!";
26 : }
27 400072 : else if (!requests_.count(seq))
28 : {
29 800144 : TLOG(TLVL_DEBUG + 36) << "Received request for sequence ID " << seq
30 400072 : << " and timestamp " << ts;
31 400072 : if (recently_completed_requests_.count(seq))
32 : {
33 0 : TLOG(TLVL_DEBUG + 36) << "Already serviced this request ( sequence ID " << seq << ")! Ignoring...";
34 : }
35 : else
36 : {
37 400072 : requests_[seq] = ts;
38 400072 : request_timing_[seq] = std::chrono::steady_clock::now();
39 : }
40 : }
41 400072 : request_cv_.notify_all();
42 400072 : }
43 :
44 1 : void artdaq::RequestBuffer::reset()
45 : {
46 1 : std::lock_guard<std::mutex> lk(request_mutex_);
47 1 : requests_.clear();
48 1 : request_timing_.clear();
49 1 : next_requests_.clear();
50 1 : recently_completed_requests_.clear();
51 1 : }
52 :
53 : /// <summary>
54 : /// Get the current requests
55 : /// </summary>
56 : /// <returns>Map relating sequence IDs to timestamps</returns>
57 49 : std::map<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> artdaq::RequestBuffer::GetRequests() const
58 : {
59 49 : std::lock_guard<std::mutex> lk(request_mutex_);
60 49 : std::map<artdaq::Fragment::sequence_id_t, Fragment::timestamp_t> out;
61 264995 : for (auto& in : requests_)
62 : {
63 264946 : out[in.first] = in.second;
64 : }
65 98 : return out;
66 49 : }
67 :
68 87 : std::map<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> artdaq::RequestBuffer::GetRequests(std::chrono::steady_clock::time_point older_than) const
69 : {
70 87 : std::lock_guard<std::mutex> lk(request_mutex_);
71 87 : std::map<artdaq::Fragment::sequence_id_t, Fragment::timestamp_t> out;
72 91665 : for (auto& in : requests_)
73 : {
74 91578 : if (request_timing_.count(in.first) && request_timing_.at(in.first) < older_than)
75 : {
76 35099 : out[in.first] = in.second;
77 : }
78 : }
79 174 : return out;
80 87 : }
81 :
82 0 : std::pair<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> artdaq::RequestBuffer::GetNextRequest()
83 : {
84 0 : std::lock_guard<std::mutex> lk(request_mutex_);
85 :
86 0 : auto it = requests_.begin();
87 0 : while (it != requests_.end() && next_requests_.count(it->first)) { ++it; }
88 :
89 0 : if (it == requests_.end())
90 : {
91 0 : return std::make_pair<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t>(0, 0);
92 : }
93 :
94 0 : next_requests_.insert(it->first);
95 0 : return *it;
96 0 : }
97 :
98 300031 : void artdaq::RequestBuffer::RemoveRequest(artdaq::Fragment::sequence_id_t reqID)
99 : {
100 600062 : TLOG(TLVL_DEBUG + 35) << "RemoveRequest: Removing request for id " << reqID;
101 300031 : std::lock_guard<std::mutex> lk(request_mutex_);
102 300031 : requests_.erase(reqID);
103 300031 : next_requests_.erase(reqID);
104 300031 : recently_completed_requests_.insert(reqID);
105 :
106 597031 : while (recently_completed_requests_.size() > recently_completed_request_history_size_)
107 : {
108 297000 : recently_completed_requests_.erase(recently_completed_requests_.begin());
109 : }
110 :
111 300031 : if (metricMan && request_timing_.count(reqID))
112 : {
113 2100217 : metricMan->sendMetric("Request Response Time", TimeUtils::GetElapsedTime(request_timing_[reqID]), "seconds", 2, MetricMode::Average);
114 : }
115 300031 : request_timing_.erase(reqID);
116 300031 : }
117 :
118 : /// <summary>
119 : /// Clear all requests from the map
120 : /// </summary>
121 :
122 0 : void artdaq::RequestBuffer::ClearRequests()
123 : {
124 0 : GetAndClearRequests();
125 0 : }
126 :
127 : /// <summary>
128 : /// Get the current requests, then clear the map
129 : /// </summary>
130 : /// <returns>Map relating sequence IDs to timestamps</returns>
131 :
132 33 : std::map<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> artdaq::RequestBuffer::GetAndClearRequests()
133 : {
134 33 : std::lock_guard<std::mutex> lk(request_mutex_);
135 33 : std::map<artdaq::Fragment::sequence_id_t, Fragment::timestamp_t> out;
136 70 : for (auto& in : requests_)
137 : {
138 37 : out[in.first] = in.second;
139 37 : recently_completed_requests_.insert(in.first);
140 : }
141 33 : while (recently_completed_requests_.size() > recently_completed_request_history_size_)
142 : {
143 0 : recently_completed_requests_.erase(recently_completed_requests_.begin());
144 : }
145 :
146 33 : next_requests_.clear();
147 33 : requests_.clear();
148 33 : request_timing_.clear();
149 66 : return out;
150 33 : }
151 :
152 : /// <summary>
153 : /// Get the number of requests currently stored in the RequestReceiver
154 : /// </summary>
155 : /// <returns>The number of requests stored in the RequestReceiver</returns>
156 :
157 182 : size_t artdaq::RequestBuffer::size()
158 : {
159 182 : std::lock_guard<std::mutex> tlk(request_mutex_);
160 364 : return requests_.size();
161 182 : }
162 :
163 : /// <summary>
164 : /// Wait for a new request message, up to the timeout given
165 : /// </summary>
166 : /// <param name="timeout_ms">Milliseconds to wait for a new request to arrive</param>
167 : /// <returns>True if any requests are present in the request map</returns>
168 :
169 100 : bool artdaq::RequestBuffer::WaitForRequests(int timeout_ms)
170 : {
171 100 : std::unique_lock<std::mutex> lk(request_mutex_); // Lock needed by wait_for
172 : // See if we have to wait at all
173 100 : if (requests_.size() > 0) return true;
174 : // If we do have to wait, check requests_.size to make sure we're not being notified spuriously
175 300 : return request_cv_.wait_for(lk, std::chrono::milliseconds(timeout_ms), [this]() { return requests_.size() > 0; });
176 100 : }
177 :
178 : /// <summary>
179 : /// Get the time a given request was received
180 : /// </summary>
181 : /// <param name="reqID">Request ID of the request</param>
182 : /// <returns>steady_clock::time_point corresponding to when the request was received</returns>
183 :
184 30 : std::chrono::steady_clock::time_point artdaq::RequestBuffer::GetRequestTime(artdaq::Fragment::sequence_id_t reqID)
185 : {
186 30 : std::lock_guard<std::mutex> lk(request_mutex_);
187 60 : return request_timing_.count(reqID) ? request_timing_[reqID] : std::chrono::steady_clock::now();
188 30 : }
|
