Line data Source code
1 : #include "artdaq/DAQdata/Globals.hh"
2 : #define TRACE_NAME (app_name + "_FragmentBuffer").c_str() // include these 2 first -
3 :
4 : #include "artdaq/DAQrate/FragmentBuffer.hh"
5 :
6 : #include <boost/exception/all.hpp>
7 : #include <boost/throw_exception.hpp>
8 :
9 : #include <iterator>
10 : #include <limits>
11 :
12 : #include "canvas/Utilities/Exception.h"
13 : #include "cetlib_except/exception.h"
14 : #include "fhiclcpp/ParameterSet.h"
15 :
16 : #include "artdaq-core/Data/ContainerFragmentLoader.hh"
17 : #include "artdaq-core/Data/Fragment.hh"
18 : #include "artdaq-core/Utilities/ExceptionHandler.hh"
19 : #include "artdaq-core/Utilities/SimpleLookupPolicy.hh"
20 : #include "artdaq-core/Utilities/TimeUtils.hh"
21 :
22 : #include <sys/poll.h>
23 : #include <algorithm>
24 : #include <fstream>
25 : #include <iomanip>
26 : #include <iostream>
27 : #include <iterator>
28 : #include "artdaq/DAQdata/TCPConnect.hh"
29 :
30 : #define TLVL_ADDFRAGMENT 32
31 : #define TLVL_CHECKSTOP 33
32 : #define TLVL_WAITFORBUFFERREADY 34
33 : #define TLVL_GETBUFFERSTATS 35
34 : #define TLVL_CHECKDATABUFFER 36
35 : #define TLVL_APPLYREQUESTS 37
36 : #define TLVL_APPLYREQUESTS_VERBOSE 38
37 : #define TLVL_SENDEMPTYFRAGMENTS 39
38 : #define TLVL_CHECKWINDOWS 40
39 : #define TLVL_EMPTYFRAGMENT 41
40 :
41 28 : artdaq::FragmentBuffer::FragmentBuffer(const fhicl::ParameterSet& ps)
42 28 : : requestBuffer_()
43 56 : , bufferModeKeepLatest_(ps.get<bool>("buffer_mode_keep_latest", false))
44 56 : , windowOffset_(ps.get<Fragment::timestamp_t>("request_window_offset", 0))
45 56 : , windowWidth_(ps.get<Fragment::timestamp_t>("request_window_width", 0))
46 56 : , staleFragmentTimeout_(ps.get<Fragment::timestamp_t>("stale_fragment_timeout", 0))
47 56 : , expectedType_(ps.get<Fragment::type_t>("expected_fragment_type", Fragment::type_t(Fragment::EmptyFragmentType)))
48 56 : , uniqueWindows_(ps.get<bool>("request_windows_are_unique", true))
49 56 : , sendMissingFragments_(ps.get<bool>("send_missing_request_fragments", true))
50 56 : , missing_request_window_timeout_us_(ps.get<size_t>("missing_request_window_timeout_us", 5000000))
51 56 : , window_close_timeout_us_(ps.get<size_t>("window_close_timeout_us", 2000000))
52 56 : , error_on_empty_(ps.get<bool>("error_on_empty_fragment", false))
53 56 : , circularDataBufferMode_(ps.get<bool>("circular_buffer_mode", false))
54 56 : , maxDataBufferDepthFragments_(ps.get<int>("data_buffer_depth_fragments", 1000))
55 56 : , maxDataBufferDepthBytes_(ps.get<size_t>("data_buffer_depth_mb", 1000) * 1024 * 1024)
56 28 : , systemFragmentCount_(0)
57 112 : , should_stop_(false)
58 : {
59 84 : auto fragment_ids = ps.get<std::vector<artdaq::Fragment::fragment_id_t>>("fragment_ids", std::vector<artdaq::Fragment::fragment_id_t>());
60 :
61 56 : TLOG(TLVL_DEBUG + 33) << "artdaq::FragmentBuffer::FragmentBuffer(ps)";
62 56 : int fragment_id = ps.get<int>("fragment_id", -99);
63 :
64 28 : if (fragment_id != -99)
65 : {
66 22 : if (fragment_ids.size() != 0)
67 : {
68 1 : auto report = "Error in FragmentBuffer: can't both define \"fragment_id\" and \"fragment_ids\" in FHiCL document";
69 3 : TLOG(TLVL_ERROR) << report;
70 3 : throw cet::exception("FragmentBufferConfig") << report;
71 : }
72 : else
73 : {
74 21 : fragment_ids.emplace_back(fragment_id);
75 : }
76 : }
77 :
78 66 : for (auto& id : fragment_ids)
79 : {
80 39 : dataBuffers_[id] = std::make_shared<DataBuffer>();
81 39 : dataBuffers_[id]->DataBufferDepthBytes = 0;
82 39 : dataBuffers_[id]->DataBufferDepthFragments = 0;
83 39 : dataBuffers_[id]->BufferFragmentKept = false;
84 : }
85 :
86 81 : std::string modeString = ps.get<std::string>("request_mode", "ignored");
87 27 : if (modeString == "single" || modeString == "Single")
88 : {
89 3 : mode_ = RequestMode::Single;
90 : }
91 24 : else if (modeString.find("buffer") != std::string::npos || modeString.find("Buffer") != std::string::npos)
92 : {
93 5 : mode_ = RequestMode::Buffer;
94 : }
95 19 : else if (modeString == "window" || modeString == "Window")
96 : {
97 14 : mode_ = RequestMode::Window;
98 : }
99 5 : else if (modeString.find("ignore") != std::string::npos || modeString.find("Ignore") != std::string::npos)
100 : {
101 3 : mode_ = RequestMode::Ignored;
102 : }
103 2 : else if (modeString.find("sequence") != std::string::npos || modeString.find("Sequence") != std::string::npos)
104 : {
105 2 : mode_ = RequestMode::SequenceID;
106 : }
107 75 : if (mode_ != RequestMode::Ignored && !ps.get<bool>("receive_requests", false))
108 : {
109 3 : TLOG(TLVL_WARNING) << "Request Mode was requested as " << modeString << ", but is being set to Ignored because \"receive_requests\" was not set to true";
110 1 : mode_ = RequestMode::Ignored;
111 : }
112 54 : TLOG(TLVL_DEBUG + 32) << "Request mode is " << printMode_();
113 32 : }
114 :
115 27 : artdaq::FragmentBuffer::~FragmentBuffer()
116 : {
117 81 : TLOG(TLVL_INFO) << "Fragment Buffer Destructor; Clearing data buffers";
118 27 : Reset(true);
119 27 : }
120 :
121 18946271 : void artdaq::FragmentBuffer::Reset(bool stop)
122 : {
123 18946271 : should_stop_ = stop;
124 37892554 : for (auto& id : dataBuffers_)
125 : {
126 18946283 : std::lock_guard<std::mutex> dlk(id.second->DataBufferMutex);
127 18946283 : id.second->DataBufferDepthBytes = 0;
128 18946283 : id.second->DataBufferDepthFragments = 0;
129 18946283 : id.second->BufferFragmentKept = false;
130 18946283 : id.second->DataBuffer.clear();
131 18946283 : }
132 :
133 : {
134 18946271 : std::lock_guard<std::mutex> lk(systemFragmentMutex_);
135 18946271 : systemFragments_.clear();
136 18946271 : systemFragmentCount_ = 0;
137 18946271 : }
138 18946271 : }
139 :
140 300054 : void artdaq::FragmentBuffer::AddFragmentsToBuffer(FragmentPtrs frags)
141 : {
142 300054 : std::unordered_map<Fragment::fragment_id_t, FragmentPtrs> frags_by_id;
143 800277 : while (!frags.empty())
144 : {
145 500223 : auto dataIter = frags.begin();
146 500223 : auto frag_id = (*dataIter)->fragmentID();
147 :
148 500223 : if (Fragment::isBroadcastFragmentType((*dataIter)->type()))
149 : {
150 0 : std::lock_guard<std::mutex> lk(systemFragmentMutex_);
151 0 : systemFragments_.emplace_back(std::move(*dataIter));
152 0 : systemFragmentCount_++;
153 0 : frags.erase(dataIter);
154 0 : continue;
155 0 : }
156 :
157 500223 : if (!dataBuffers_.count(frag_id))
158 : {
159 0 : TLOG(TLVL_ERROR) << "Received Fragment with Fragment ID " << frag_id << ", which is not in the declared Fragment IDs list!";
160 0 : throw cet::exception("FragmentIDs") << "Received Fragment with Fragment ID " << frag_id << ", which is not in the declared Fragment IDs list!";
161 : }
162 :
163 500223 : frags_by_id[frag_id].emplace_back(std::move(*dataIter));
164 500223 : frags.erase(dataIter);
165 : }
166 :
167 300054 : auto type_it = frags_by_id.begin();
168 600141 : while (type_it != frags_by_id.end())
169 : {
170 300087 : auto frag_id = type_it->first;
171 :
172 300087 : waitForDataBufferReady(frag_id);
173 300087 : auto dataBuffer = dataBuffers_[frag_id];
174 300087 : std::lock_guard<std::mutex> dlk(dataBuffer->DataBufferMutex);
175 300087 : switch (mode_)
176 : {
177 15 : case RequestMode::Single: {
178 15 : auto dataIter = type_it->second.rbegin();
179 30 : TLOG(TLVL_ADDFRAGMENT) << "Adding Fragment with Fragment ID " << frag_id << ", Sequence ID " << (*dataIter)->sequenceID() << ", and Timestamp " << (*dataIter)->timestamp() << " to buffer";
180 15 : dataBuffer->DataBuffer.clear();
181 15 : dataBuffer->DataBufferDepthBytes = (*dataIter)->sizeBytes();
182 15 : dataBuffer->DataBuffer.emplace_back(std::move(*dataIter));
183 15 : dataBuffer->DataBufferDepthFragments = 1;
184 15 : type_it->second.clear();
185 : }
186 15 : break;
187 300072 : case RequestMode::Buffer:
188 : case RequestMode::Ignored:
189 : case RequestMode::Window:
190 : case RequestMode::SequenceID:
191 : default:
192 800274 : while (!type_it->second.empty())
193 : {
194 500202 : auto dataIter = type_it->second.begin();
195 1000404 : TLOG(TLVL_ADDFRAGMENT) << "Adding Fragment with Fragment ID " << frag_id << ", Sequence ID " << (*dataIter)->sequenceID() << ", and Timestamp " << (*dataIter)->timestamp() << " to buffer";
196 :
197 500202 : dataBuffer->DataBufferDepthBytes += (*dataIter)->sizeBytes();
198 500202 : dataBuffer->DataBuffer.emplace_back(std::move(*dataIter));
199 500202 : type_it->second.erase(dataIter);
200 : }
201 300072 : dataBuffer->DataBufferDepthFragments = dataBuffer->DataBuffer.size();
202 300072 : break;
203 : }
204 300087 : getDataBufferStats(frag_id);
205 300087 : ++type_it;
206 300087 : }
207 300054 : dataCondition_.notify_all();
208 300054 : }
209 :
210 371 : bool artdaq::FragmentBuffer::check_stop()
211 : {
212 742 : TLOG(TLVL_CHECKSTOP) << "CFG::check_stop: should_stop=" << should_stop_.load();
213 :
214 371 : if (!should_stop_.load()) return false;
215 4 : if (mode_ == RequestMode::Ignored)
216 : {
217 1 : return true;
218 : }
219 :
220 3 : if (requestBuffer_ != nullptr)
221 : {
222 : // check_stop returns true if the CFG should stop. We should wait for the Request Buffer to report Request Receiver stopped before stopping.
223 6 : TLOG(TLVL_DEBUG + 32) << "should_stop is true, requestBuffer_->isRunning() is " << std::boolalpha << requestBuffer_->isRunning();
224 3 : if (!requestBuffer_->isRunning())
225 : {
226 1 : return true;
227 : }
228 : }
229 2 : return false;
230 : }
231 :
232 0 : std::string artdaq::FragmentBuffer::printMode_()
233 : {
234 0 : switch (mode_)
235 : {
236 0 : case RequestMode::Single:
237 0 : return "Single";
238 0 : case RequestMode::Buffer:
239 0 : return "Buffer";
240 0 : case RequestMode::Window:
241 0 : return "Window";
242 0 : case RequestMode::Ignored:
243 0 : return "Ignored";
244 0 : case RequestMode::SequenceID:
245 0 : return "SequenceID";
246 : }
247 :
248 0 : return "ERROR";
249 : }
250 :
251 305 : size_t artdaq::FragmentBuffer::dataBufferFragmentCount_()
252 : {
253 305 : size_t count = 0;
254 612 : for (auto& id : dataBuffers_) count += id.second->DataBufferDepthFragments;
255 305 : count += systemFragmentCount_.load();
256 305 : return count;
257 : }
258 :
259 300087 : bool artdaq::FragmentBuffer::waitForDataBufferReady(Fragment::fragment_id_t id)
260 : {
261 300087 : if (!dataBuffers_.count(id))
262 : {
263 0 : TLOG(TLVL_ERROR) << "DataBufferError: "
264 0 : << "Error in FragmentBuffer: Cannot wait for data buffer for ID " << id << " because it does not exist!";
265 0 : throw cet::exception("DataBufferError") << "Error in FragmentBuffer: Cannot wait for data buffer for ID " << id << " because it does not exist!";
266 : }
267 300087 : auto startwait = std::chrono::steady_clock::now();
268 300087 : auto first = true;
269 300087 : auto lastwaittime = 0ULL;
270 300087 : auto dataBuffer = dataBuffers_[id];
271 :
272 303336 : while (dataBufferIsTooLarge(id))
273 : {
274 3249 : if (!circularDataBufferMode_)
275 : {
276 3249 : if (should_stop_.load())
277 : {
278 0 : TLOG(TLVL_DEBUG + 32) << "Run ended while waiting for buffer to shrink!";
279 0 : getDataBufferStats(id);
280 0 : dataCondition_.notify_all();
281 0 : return false;
282 : }
283 3249 : auto waittime = TimeUtils::GetElapsedTimeMilliseconds(startwait);
284 :
285 3249 : if (first || (waittime != lastwaittime && waittime % 1000 == 0))
286 : {
287 3249 : std::lock_guard<std::mutex> lk(dataBuffer->DataBufferMutex);
288 3249 : if (dataBufferIsTooLarge(id))
289 : {
290 4290 : TLOG(TLVL_WARNING) << "Bad Omen: Data Buffer has exceeded its size limits. "
291 1430 : << "(frag_id=" << id
292 1430 : << ", frags=" << dataBuffer->DataBufferDepthFragments << "/" << maxDataBufferDepthFragments_
293 1430 : << ", szB=" << dataBuffer->DataBufferDepthBytes << "/" << maxDataBufferDepthBytes_ << ")"
294 2860 : << ", timestamps=" << dataBuffer->DataBuffer.front()->timestamp() << "-" << dataBuffer->DataBuffer.back()->timestamp();
295 2860 : TLOG(TLVL_DEBUG + 33) << "Bad Omen: Possible causes include requests not getting through or Ignored-mode BR issues";
296 :
297 1430 : if (metricMan)
298 : {
299 10010 : metricMan->sendMetric("Bad Omen wait time", waittime / 1000.0, "s", 1, MetricMode::LastPoint);
300 : }
301 : }
302 3249 : first = false;
303 3249 : }
304 3249 : if (waittime % 5 && waittime != lastwaittime)
305 : {
306 0 : TLOG(TLVL_WAITFORBUFFERREADY) << "getDataLoop: Data Retreival paused for " << waittime << " ms waiting for data buffer to drain";
307 : }
308 3249 : lastwaittime = waittime;
309 3249 : usleep(1000);
310 : }
311 : else
312 : {
313 0 : std::lock_guard<std::mutex> lk(dataBuffer->DataBufferMutex);
314 0 : if (dataBufferIsTooLarge(id))
315 : {
316 0 : auto begin = dataBuffer->DataBuffer.begin();
317 0 : if (begin == dataBuffer->DataBuffer.end())
318 : {
319 0 : TLOG(TLVL_WARNING) << "Data buffer is reported as too large, but doesn't contain any Fragments! Possible corrupt memory!";
320 0 : continue;
321 0 : }
322 0 : if (*begin)
323 : {
324 0 : TLOG(TLVL_WAITFORBUFFERREADY) << "waitForDataBufferReady: Dropping Fragment with timestamp " << (*begin)->timestamp() << " from data buffer (Buffer over-size, circular data buffer mode)";
325 :
326 0 : dataBuffer->DataBufferDepthBytes -= (*begin)->sizeBytes();
327 0 : dataBuffer->DataBuffer.erase(begin);
328 0 : dataBuffer->DataBufferDepthFragments = dataBuffer->DataBuffer.size();
329 0 : dataBuffer->BufferFragmentKept = false; // If any Fragments are removed from data buffer, then we know we don't have to ignore the first one anymore
330 : }
331 : }
332 0 : }
333 : }
334 300087 : return true;
335 300087 : }
336 :
337 356740 : bool artdaq::FragmentBuffer::dataBufferIsTooLarge(Fragment::fragment_id_t id)
338 : {
339 356740 : if (!dataBuffers_.count(id))
340 : {
341 0 : TLOG(TLVL_ERROR) << "DataBufferError: "
342 0 : << "Error in FragmentBuffer: Cannot check size of data buffer for ID " << id << " because it does not exist!";
343 0 : throw cet::exception("DataBufferError") << "Error in FragmentBuffer: Cannot check size of data buffer for ID " << id << " because it does not exist!";
344 : }
345 356740 : auto dataBuffer = dataBuffers_[id];
346 1015477 : return (maxDataBufferDepthFragments_ > 0 && dataBuffer->DataBufferDepthFragments.load() > maxDataBufferDepthFragments_) ||
347 960734 : (maxDataBufferDepthBytes_ > 0 && dataBuffer->DataBufferDepthBytes.load() > maxDataBufferDepthBytes_);
348 356740 : }
349 :
350 300232 : void artdaq::FragmentBuffer::getDataBufferStats(Fragment::fragment_id_t id)
351 : {
352 300232 : if (!dataBuffers_.count(id))
353 : {
354 0 : TLOG(TLVL_ERROR) << "DataBufferError: "
355 0 : << "Error in FragmentBuffer: Cannot get stats of data buffer for ID " << id << " because it does not exist!";
356 0 : throw cet::exception("DataBufferError") << "Error in FragmentBuffer: Cannot get stats of data buffer for ID " << id << " because it does not exist!";
357 : }
358 300232 : auto dataBuffer = dataBuffers_[id];
359 :
360 300232 : if (metricMan)
361 : {
362 600464 : TLOG(TLVL_GETBUFFERSTATS) << "getDataBufferStats: Sending Metrics";
363 2101624 : metricMan->sendMetric("Buffer Depth Fragments", dataBuffer->DataBufferDepthFragments.load(), "fragments", 1, MetricMode::LastPoint);
364 2101624 : metricMan->sendMetric("Buffer Depth Bytes", dataBuffer->DataBufferDepthBytes.load(), "bytes", 1, MetricMode::LastPoint);
365 :
366 300232 : auto bufferDepthFragmentsPercent = dataBuffer->DataBufferDepthFragments.load() * 100 / static_cast<double>(maxDataBufferDepthFragments_);
367 300232 : auto bufferDepthBytesPercent = dataBuffer->DataBufferDepthBytes.load() * 100 / static_cast<double>(maxDataBufferDepthBytes_);
368 1801392 : metricMan->sendMetric("Fragment Buffer Full %Fragments", bufferDepthFragmentsPercent, "%", 3, MetricMode::LastPoint);
369 1801392 : metricMan->sendMetric("Fragment Buffer Full %Bytes", bufferDepthBytesPercent, "%", 3, MetricMode::LastPoint);
370 2101624 : metricMan->sendMetric("Fragment Buffer Full %", bufferDepthFragmentsPercent > bufferDepthBytesPercent ? bufferDepthFragmentsPercent : bufferDepthBytesPercent, "%", 1, MetricMode::LastPoint);
371 : }
372 600464 : TLOG(TLVL_GETBUFFERSTATS) << "getDataBufferStats: frags=" << dataBuffer->DataBufferDepthFragments.load() << "/" << maxDataBufferDepthFragments_
373 300232 : << ", sz=" << dataBuffer->DataBufferDepthBytes.load() << "/" << maxDataBufferDepthBytes_;
374 300232 : }
375 :
376 0 : std::string artdaq::FragmentBuffer::getStatReport()
377 : {
378 0 : std::ostringstream oss;
379 :
380 0 : for (auto& it : dataBuffers_)
381 : {
382 0 : Fragment::fragment_id_t id = it.first;
383 0 : if (!dataBuffers_.count(id))
384 : {
385 0 : TLOG(TLVL_ERROR) << "DataBufferError: "
386 0 : << "Error in FragmentBuffer: Cannot get stats of data buffer for ID "
387 0 : << id << " because it does not exist!";
388 0 : throw cet::exception("DataBufferError") << "Error in FragmentBuffer: Cannot get stats of data buffer for ID "
389 0 : << id << " because it does not exist!";
390 : }
391 :
392 0 : auto dataBuffer = dataBuffers_[id];
393 :
394 0 : TLOG(TLVL_GETBUFFERSTATS) << "getDataBufferStats: Sending Metrics";
395 :
396 0 : int nf = dataBuffer->DataBufferDepthFragments.load();
397 0 : int nb = dataBuffer->DataBufferDepthBytes.load();
398 0 : oss << std::endl
399 0 : << "fragment_id:" << id << " nfragments:" << nf << " nbytes:" << nb
400 0 : << " max_nf:" << maxDataBufferDepthFragments_ << " max_nb:" << maxDataBufferDepthBytes_;
401 :
402 0 : TLOG(TLVL_GETBUFFERSTATS) << "getDataBufferStats: frags=" << dataBuffer->DataBufferDepthFragments.load() << "/" << maxDataBufferDepthFragments_
403 0 : << ", sz=" << dataBuffer->DataBufferDepthBytes.load() << "/" << maxDataBufferDepthBytes_;
404 0 : }
405 :
406 0 : return oss.str();
407 0 : }
408 :
409 238 : void artdaq::FragmentBuffer::checkDataBuffer(Fragment::fragment_id_t id)
410 : {
411 238 : if (!dataBuffers_.count(id))
412 : {
413 0 : TLOG(TLVL_ERROR) << "DataBufferError: "
414 0 : << "Error in FragmentBuffer: Cannot check data buffer for ID " << id << " because it does not exist!";
415 0 : throw cet::exception("DataBufferError") << "Error in FragmentBuffer: Cannot check data buffer for ID " << id << " because it does not exist!";
416 : }
417 :
418 238 : if (dataBuffers_[id]->DataBufferDepthFragments > 0 && mode_ != RequestMode::Single && mode_ != RequestMode::Ignored)
419 : {
420 91 : auto dataBuffer = dataBuffers_[id];
421 91 : std::lock_guard<std::mutex> lk(dataBuffer->DataBufferMutex);
422 :
423 : // Eliminate extra fragments
424 50155 : while (dataBufferIsTooLarge(id))
425 : {
426 50064 : auto begin = dataBuffer->DataBuffer.begin();
427 100128 : TLOG(TLVL_CHECKDATABUFFER) << "checkDataBuffer: Dropping Fragment with timestamp " << (*begin)->timestamp() << " from data buffer (Buffer over-size)";
428 50064 : dataBuffer->DataBufferDepthBytes -= (*begin)->sizeBytes();
429 50064 : dataBuffer->DataBuffer.erase(begin);
430 50064 : dataBuffer->DataBufferDepthFragments = dataBuffer->DataBuffer.size();
431 50064 : dataBuffer->BufferFragmentKept = false; // If any Fragments are removed from data buffer, then we know we don't have to ignore the first one anymore
432 : }
433 :
434 182 : TLOG(TLVL_CHECKDATABUFFER) << "DataBufferDepthFragments is " << dataBuffer->DataBufferDepthFragments << ", DataBuffer.size is " << dataBuffer->DataBuffer.size();
435 91 : if (dataBuffer->DataBufferDepthFragments > 0 && staleFragmentTimeout_ > 0)
436 : {
437 0 : TLOG(TLVL_CHECKDATABUFFER) << "Determining if Fragments can be dropped from data buffer";
438 0 : Fragment::timestamp_t last = dataBuffer->DataBuffer.back()->timestamp();
439 0 : Fragment::timestamp_t min = last > staleFragmentTimeout_ ? last - staleFragmentTimeout_ : 0;
440 0 : for (auto it = dataBuffer->DataBuffer.begin(); it != dataBuffer->DataBuffer.end();)
441 : {
442 0 : if ((*it)->timestamp() < min)
443 : {
444 0 : TLOG(TLVL_CHECKDATABUFFER) << "checkDataBuffer: Dropping Fragment with timestamp " << (*it)->timestamp() << " from data buffer (timeout=" << staleFragmentTimeout_ << ", min=" << min << ")";
445 0 : dataBuffer->DataBufferDepthBytes -= (*it)->sizeBytes();
446 0 : dataBuffer->BufferFragmentKept = false; // If any Fragments are removed from data buffer, then we know we don't have to ignore the first one anymore
447 0 : it = dataBuffer->DataBuffer.erase(it);
448 0 : dataBuffer->DataBufferDepthFragments = dataBuffer->DataBuffer.size();
449 : }
450 : else
451 : {
452 0 : break;
453 : }
454 : }
455 : }
456 91 : }
457 238 : }
458 :
459 5 : void artdaq::FragmentBuffer::applyRequestsIgnoredMode(artdaq::FragmentPtrs& frags)
460 : {
461 : // dataBuffersMutex_ is held by calling function
462 : // We just copy everything that's here into the output.
463 10 : TLOG(TLVL_APPLYREQUESTS) << "Mode is Ignored; Copying data to output";
464 12 : for (auto& id : dataBuffers_)
465 : {
466 7 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
467 7 : std::move(id.second->DataBuffer.begin(), id.second->DataBuffer.end(), std::inserter(frags, frags.end()));
468 7 : id.second->DataBufferDepthBytes = 0;
469 7 : id.second->DataBufferDepthFragments = 0;
470 7 : id.second->BufferFragmentKept = false;
471 7 : id.second->DataBuffer.clear();
472 7 : }
473 5 : }
474 :
475 13 : void artdaq::FragmentBuffer::applyRequestsSingleMode(artdaq::FragmentPtrs& frags)
476 : {
477 : // We only care about the latest request received. Send empties for all others.
478 13 : auto requests = requestBuffer_->GetAndClearRequests();
479 13 : if (requests.size() == 0) return;
480 :
481 12 : auto last_request = requests.rbegin();
482 :
483 34 : for (auto& id : dataBuffers_)
484 : {
485 48 : for (auto& request : requests)
486 : {
487 26 : if (request.first != last_request->first)
488 : {
489 8 : sendEmptyFragment(frags, request.first, id.first, "Single Mode stale request");
490 : }
491 : }
492 22 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
493 22 : if (id.second->DataBufferDepthFragments > 0)
494 : {
495 22 : assert(id.second->DataBufferDepthFragments == 1);
496 44 : TLOG(TLVL_APPLYREQUESTS) << "Mode is Single; Sending copy of most recent Fragment (timestamp=" << (*id.second->DataBuffer.begin())->timestamp() << ")";
497 44 : for (auto& fragptr : id.second->DataBuffer)
498 : {
499 : // Return the latest data point
500 22 : auto frag = fragptr.get();
501 22 : auto newfrag = std::unique_ptr<artdaq::Fragment>(new Fragment(last_request->first, frag->fragmentID(), frag->type()));
502 22 : newfrag->resize(frag->size() - detail::RawFragmentHeader::num_words());
503 22 : memcpy(newfrag->headerAddress(), frag->headerAddress(), frag->sizeBytes());
504 22 : newfrag->setTimestamp(last_request->second);
505 22 : newfrag->setSequenceID(last_request->first);
506 22 : frags.push_back(std::move(newfrag));
507 22 : }
508 : }
509 : else
510 : {
511 0 : sendEmptyFragment(frags, last_request->first, id.first, "No data for");
512 : }
513 22 : }
514 13 : }
515 :
516 20 : void artdaq::FragmentBuffer::applyRequestsBufferMode(artdaq::FragmentPtrs& frags)
517 : {
518 : // We only care about the latest request received. Send empties for all others.
519 20 : auto requests = requestBuffer_->GetAndClearRequests();
520 20 : if (requests.size() == 0) return;
521 :
522 20 : auto last_request = requests.rbegin();
523 :
524 56 : for (auto& id : dataBuffers_)
525 : {
526 77 : for (auto& request : requests)
527 : {
528 41 : if (request.first != last_request->first)
529 : {
530 10 : sendEmptyFragment(frags, request.first, id.first, "Buffer Mode stale request");
531 : }
532 : }
533 :
534 72 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsBufferMode: Creating ContainerFragment for Buffered Fragments (SeqID " << last_request->first << ")";
535 36 : frags.emplace_back(new artdaq::Fragment(last_request->first, id.first, artdaq::Fragment::ContainerFragmentType));
536 36 : frags.back()->setTimestamp(last_request->second);
537 36 : ContainerFragmentLoader cfl(*frags.back());
538 36 : cfl.set_missing_data(false); // Buffer mode is never missing data, even if there IS no data.
539 :
540 : // If we kept a Fragment from the previous iteration, but more data has arrived, discard it
541 36 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
542 36 : if (id.second->BufferFragmentKept && id.second->DataBufferDepthFragments > 1)
543 : {
544 1 : id.second->DataBufferDepthBytes -= id.second->DataBuffer.front()->sizeBytes();
545 1 : id.second->DataBuffer.erase(id.second->DataBuffer.begin());
546 1 : id.second->DataBufferDepthFragments = id.second->DataBuffer.size();
547 : }
548 :
549 : // Buffer mode TFGs should simply copy out the whole dataBuffer_ into a ContainerFragment
550 36 : FragmentPtrs fragsToAdd;
551 36 : std::move(id.second->DataBuffer.begin(), --id.second->DataBuffer.end(), std::back_inserter(fragsToAdd));
552 36 : id.second->DataBuffer.erase(id.second->DataBuffer.begin(), --id.second->DataBuffer.end());
553 :
554 36 : if (fragsToAdd.size() > 0)
555 : {
556 34 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsBufferMode: Adding " << fragsToAdd.size() << " Fragments to Container (SeqID " << last_request->first << ")";
557 17 : cfl.addFragments(fragsToAdd);
558 : }
559 : else
560 : {
561 38 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsBufferMode: No Fragments to add (SeqID " << last_request->first << ")";
562 : }
563 :
564 36 : if (id.second->DataBuffer.size() == 1)
565 : {
566 56 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsBufferMode: Adding Fragment with timestamp " << id.second->DataBuffer.front()->timestamp() << " to Container with sequence ID " << last_request->first;
567 28 : cfl.addFragment(id.second->DataBuffer.front());
568 28 : if (bufferModeKeepLatest_)
569 : {
570 4 : id.second->BufferFragmentKept = true;
571 4 : id.second->DataBufferDepthBytes = id.second->DataBuffer.front()->sizeBytes();
572 4 : id.second->DataBufferDepthFragments = id.second->DataBuffer.size(); // 1
573 : }
574 : else
575 : {
576 24 : id.second->DataBuffer.clear();
577 24 : id.second->BufferFragmentKept = false;
578 24 : id.second->DataBufferDepthBytes = 0;
579 24 : id.second->DataBufferDepthFragments = 0;
580 : }
581 : }
582 36 : }
583 20 : }
584 :
585 264954 : void artdaq::FragmentBuffer::applyRequestsWindowMode_CheckAndFillDataBuffer(artdaq::FragmentPtrs& frags, artdaq::Fragment::fragment_id_t id, artdaq::Fragment::sequence_id_t seq, artdaq::Fragment::timestamp_t ts)
586 : {
587 264954 : auto dataBuffer = dataBuffers_[id];
588 :
589 529908 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsWindowMode_CheckAndFillDataBuffer: Checking that data exists for request window " << seq;
590 264954 : Fragment::timestamp_t min = ts > windowOffset_ ? ts - windowOffset_ : 0;
591 264954 : Fragment::timestamp_t max = ts + windowWidth_ > windowOffset_ ? ts + windowWidth_ - windowOffset_ : 1;
592 :
593 529908 : TLOG(TLVL_APPLYREQUESTS) << "ApplyRequestsWindowsMode_CheckAndFillDataBuffer: min is " << min << ", max is " << max
594 0 : << " and first/last points in buffer are " << (dataBuffer->DataBufferDepthFragments > 0 ? dataBuffer->DataBuffer.front()->timestamp() : 0)
595 0 : << "/" << (dataBuffer->DataBufferDepthFragments > 0 ? dataBuffer->DataBuffer.back()->timestamp() : 0)
596 0 : << " (sz=" << dataBuffer->DataBufferDepthFragments << " [" << dataBuffer->DataBufferDepthBytes.load()
597 264954 : << "/" << maxDataBufferDepthBytes_ << "])";
598 264954 : bool windowClosed = dataBuffer->DataBufferDepthFragments > 0 && dataBuffer->DataBuffer.back()->timestamp() >= max;
599 264954 : bool windowTimeout = !windowClosed && TimeUtils::GetElapsedTimeMicroseconds(requestBuffer_->GetRequestTime(seq)) > window_close_timeout_us_;
600 264954 : if (windowTimeout)
601 : {
602 21 : TLOG(TLVL_WARNING) << "applyRequestsWindowMode_CheckAndFillDataBuffer: A timeout occurred waiting for data to close the request window ({" << min << "-" << max
603 7 : << "}, buffer={" << (dataBuffer->DataBufferDepthFragments > 0 ? dataBuffer->DataBuffer.front()->timestamp() : 0) << "-"
604 7 : << (dataBuffer->DataBufferDepthFragments > 0 ? dataBuffer->DataBuffer.back()->timestamp() : 0)
605 7 : << "} ). Time waiting: "
606 14 : << TimeUtils::GetElapsedTimeMicroseconds(requestBuffer_->GetRequestTime(seq)) << " us "
607 14 : << "(> " << window_close_timeout_us_ << " us).";
608 : }
609 264954 : if (windowClosed || windowTimeout)
610 : {
611 529876 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsWindowMode_CheckAndFillDataBuffer: Creating ContainerFragment for Window-requested Fragments (SeqID " << seq << ")";
612 264938 : frags.emplace_back(new artdaq::Fragment(seq, id, artdaq::Fragment::ContainerFragmentType));
613 264938 : frags.back()->setTimestamp(ts);
614 264938 : ContainerFragmentLoader cfl(*frags.back());
615 :
616 : // In the spirit of NOvA's MegaPool: (RS = Request start (min), RE = Request End (max))
617 : // --- | Buffer Start | --- | Buffer End | ---
618 : // 1. RS RE | | | |
619 : // 2. RS | | RE | |
620 : // 3. RS | | | | RE
621 : // 4. | | RS RE | |
622 : // 5. | | RS | | RE
623 : // 6. | | | | RS RE
624 : //
625 : // If RE (or RS) is after the end of the buffer, we wait for window_close_timeout_us_. If we're here, then that means that windowClosed is false, and the missing_data flag should be set.
626 : // If RS (or RE) is before the start of the buffer, then missing_data should be set to true, as data is assumed to arrive in the buffer in timestamp order
627 : // If the dataBuffer has size 0, then windowClosed will be false
628 264938 : if (!windowClosed || (dataBuffer->DataBufferDepthFragments > 0 && dataBuffer->DataBuffer.front()->timestamp() > min))
629 : {
630 100028 : TLOG(TLVL_DEBUG + 32) << "applyRequestsWindowMode_CheckAndFillDataBuffer: Request window starts before and/or ends after the current data buffer, setting ContainerFragment's missing_data flag!"
631 0 : << " (requestWindowRange=[" << min << "," << max << "], "
632 0 : << "buffer={" << (dataBuffer->DataBufferDepthFragments > 0 ? dataBuffer->DataBuffer.front()->timestamp() : 0) << "-"
633 50014 : << (dataBuffer->DataBufferDepthFragments > 0 ? dataBuffer->DataBuffer.back()->timestamp() : 0) << "} (SeqID " << seq << ")";
634 50014 : cfl.set_missing_data(true);
635 : }
636 :
637 264938 : auto it = dataBuffer->DataBuffer.begin();
638 : // Likely that it will be closer to the end...
639 264938 : if (windowTimeout)
640 : {
641 7 : it = dataBuffer->DataBuffer.end();
642 7 : --it;
643 8 : while (it != dataBuffer->DataBuffer.begin())
644 : {
645 2 : if ((*it)->timestamp() < min)
646 : {
647 1 : break;
648 : }
649 1 : --it;
650 : }
651 : }
652 :
653 264938 : FragmentPtrs fragsToAdd;
654 : // Do a little bit more work to decide which fragments to send for a given request
655 2227546560 : for (; it != dataBuffer->DataBuffer.end();)
656 : {
657 2227546539 : Fragment::timestamp_t fragT = (*it)->timestamp();
658 2227546539 : if (fragT < min)
659 : {
660 2227066683 : ++it;
661 2227066683 : continue;
662 : }
663 479856 : if (fragT > max || (fragT == max && windowWidth_ > 0))
664 : {
665 : break;
666 : }
667 :
668 429878 : TLOG(TLVL_APPLYREQUESTS_VERBOSE) << "applyRequestsWindowMode_CheckAndFillDataBuffer: Adding Fragment with timestamp " << (*it)->timestamp() << " to Container (SeqID " << seq << ")";
669 214939 : if (uniqueWindows_)
670 : {
671 214939 : dataBuffer->DataBufferDepthBytes -= (*it)->sizeBytes();
672 214939 : fragsToAdd.emplace_back(std::move(*it));
673 214939 : it = dataBuffer->DataBuffer.erase(it);
674 : }
675 : else
676 : {
677 0 : fragsToAdd.emplace_back(it->get());
678 0 : ++it;
679 : }
680 : }
681 :
682 264938 : if (fragsToAdd.size() > 0)
683 : {
684 429856 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsWindowMode_CheckAndFillDataBuffer: Adding " << fragsToAdd.size() << " Fragments to Container (SeqID " << seq << ")";
685 214928 : cfl.addFragments(fragsToAdd);
686 :
687 : // Don't delete Fragments which are still in the Fragment buffer
688 214928 : if (!uniqueWindows_)
689 : {
690 0 : for (auto& frag : fragsToAdd)
691 : {
692 0 : frag.release();
693 : }
694 : }
695 214928 : fragsToAdd.clear();
696 : }
697 : else
698 : {
699 100020 : TLOG(error_on_empty_ ? TLVL_ERROR : TLVL_APPLYREQUESTS) << "applyRequestsWindowMode_CheckAndFillDataBuffer: No Fragments match request (SeqID " << seq << ", window " << min << " - " << max << ")";
700 : }
701 :
702 264938 : dataBuffer->DataBufferDepthFragments = dataBuffer->DataBuffer.size();
703 264938 : dataBuffer->WindowsSent.insert(seq);
704 264938 : }
705 264954 : }
706 :
707 39 : void artdaq::FragmentBuffer::applyRequestsWindowMode(artdaq::FragmentPtrs& frags)
708 : {
709 78 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsWindowMode BEGIN";
710 :
711 39 : auto requests = requestBuffer_->GetRequests();
712 :
713 78 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsWindowMode: Starting request processing for " << requests.size() << " requests";
714 264975 : for (auto req = requests.begin(); req != requests.end();)
715 : {
716 529872 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsWindowMode: processing request with sequence ID " << req->first << ", timestamp " << req->second;
717 :
718 264936 : auto ts = req->second;
719 264936 : if (ts == Fragment::InvalidTimestamp)
720 : {
721 0 : TLOG(TLVL_ERROR) << "applyRequestsWindowMode: Received InvalidTimestamp in request " << req->first << ", cannot apply! Check that push-mode BRs are filling appropriate timestamps in their Fragments!";
722 0 : req = requests.erase(req);
723 0 : continue;
724 0 : }
725 :
726 529892 : for (auto& id : dataBuffers_)
727 : {
728 264956 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
729 264956 : if (!id.second->WindowsSent.count(req->first))
730 : {
731 264954 : applyRequestsWindowMode_CheckAndFillDataBuffer(frags, id.first, req->first, req->second);
732 : }
733 264956 : }
734 264936 : checkSentWindows(req->first);
735 264936 : ++req;
736 : }
737 :
738 : // Check sent windows for requests that can be removed
739 39 : std::set<artdaq::Fragment::sequence_id_t> seqs;
740 98 : for (auto& id : dataBuffers_)
741 : {
742 59 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
743 61 : for (auto& seq : id.second->WindowsSent)
744 : {
745 2 : seqs.insert(seq);
746 : }
747 59 : }
748 40 : for (auto& seq : seqs)
749 : {
750 1 : checkSentWindows(seq);
751 : }
752 39 : }
753 :
754 10 : void artdaq::FragmentBuffer::applyRequestsSequenceIDMode(artdaq::FragmentPtrs& frags)
755 : {
756 20 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsSequenceIDMode BEGIN";
757 :
758 10 : auto requests = requestBuffer_->GetRequests();
759 :
760 20 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsSequenceIDMode: Starting request processing";
761 20 : for (auto req = requests.begin(); req != requests.end();)
762 : {
763 20 : TLOG(TLVL_APPLYREQUESTS) << "applyRequestsSequenceIDMode: Checking that data exists for request SequenceID " << req->first;
764 :
765 30 : for (auto& id : dataBuffers_)
766 : {
767 20 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
768 20 : if (!id.second->WindowsSent.count(req->first))
769 : {
770 40 : TLOG(TLVL_APPLYREQUESTS_VERBOSE) << "Searching id " << id.first << " for Fragments with Sequence ID " << req->first;
771 40 : for (auto it = id.second->DataBuffer.begin(); it != id.second->DataBuffer.end();)
772 : {
773 20 : auto seq = (*it)->sequenceID();
774 40 : TLOG(TLVL_APPLYREQUESTS_VERBOSE) << "applyRequestsSequenceIDMode: Fragment SeqID " << seq << ", request ID " << req->first;
775 20 : if (seq == req->first)
776 : {
777 32 : TLOG(TLVL_APPLYREQUESTS_VERBOSE) << "applyRequestsSequenceIDMode: Adding Fragment to output";
778 16 : id.second->WindowsSent.insert(req->first);
779 16 : id.second->DataBufferDepthBytes -= (*it)->sizeBytes();
780 16 : frags.push_back(std::move(*it));
781 16 : it = id.second->DataBuffer.erase(it);
782 16 : id.second->DataBufferDepthFragments = id.second->DataBuffer.size();
783 : }
784 : else
785 : {
786 4 : ++it;
787 : }
788 : }
789 : }
790 20 : }
791 10 : checkSentWindows(req->first);
792 10 : ++req;
793 : }
794 :
795 : // Check sent windows for requests that can be removed
796 10 : std::set<artdaq::Fragment::sequence_id_t> seqs;
797 30 : for (auto& id : dataBuffers_)
798 : {
799 20 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
800 20 : for (auto& seq : id.second->WindowsSent)
801 : {
802 0 : seqs.insert(seq);
803 : }
804 20 : }
805 10 : for (auto& seq : seqs)
806 : {
807 0 : checkSentWindows(seq);
808 : }
809 10 : }
810 :
811 89 : bool artdaq::FragmentBuffer::applyRequests(artdaq::FragmentPtrs& frags)
812 : {
813 89 : if (check_stop())
814 : {
815 2 : return false;
816 : }
817 :
818 : // Wait for data, if in ignored mode, or a request otherwise
819 87 : if (mode_ == RequestMode::Ignored)
820 : {
821 5 : auto start_time = std::chrono::steady_clock::now();
822 105 : while (dataBufferFragmentCount_() == 0 && TimeUtils::GetElapsedTime(start_time) < 1.0)
823 : {
824 100 : if (check_stop()) return false;
825 100 : std::unique_lock<std::mutex> lock(dataConditionMutex_);
826 300 : dataCondition_.wait_for(lock, std::chrono::milliseconds(10), [this]() { return dataBufferFragmentCount_() > 0; });
827 100 : }
828 : }
829 82 : else if (requestBuffer_ == nullptr)
830 : {
831 0 : TLOG(TLVL_ERROR) << "Request Buffer must be set (via SetRequestBuffer) before applyRequests/getData can be called!";
832 0 : return false;
833 : }
834 : else
835 : {
836 82 : if ((check_stop() && requestBuffer_->size() == 0)) return false;
837 :
838 82 : std::unique_lock<std::mutex> lock(dataConditionMutex_);
839 82 : dataCondition_.wait_for(lock, std::chrono::milliseconds(10));
840 :
841 82 : checkDataBuffers();
842 :
843 : // Wait up to 1000 ms for a request...
844 82 : auto counter = 0;
845 :
846 182 : while (requestBuffer_->size() == 0 && counter < 100)
847 : {
848 100 : if (check_stop()) return false;
849 :
850 100 : checkDataBuffers();
851 :
852 100 : requestBuffer_->WaitForRequests(10); // milliseconds
853 100 : counter++;
854 : }
855 82 : }
856 :
857 174 : if (systemFragmentCount_.load() > 0)
858 : {
859 0 : std::lock_guard<std::mutex> lk(systemFragmentMutex_);
860 0 : TLOG(TLVL_INFO) << "Copying " << systemFragmentCount_.load() << " System Fragments into output";
861 :
862 0 : std::move(systemFragments_.begin(), systemFragments_.end(), std::inserter(frags, frags.end()));
863 0 : systemFragments_.clear();
864 0 : systemFragmentCount_ = 0;
865 0 : }
866 :
867 87 : switch (mode_)
868 : {
869 13 : case RequestMode::Single:
870 13 : applyRequestsSingleMode(frags);
871 13 : break;
872 39 : case RequestMode::Window:
873 39 : applyRequestsWindowMode(frags);
874 39 : break;
875 20 : case RequestMode::Buffer:
876 20 : applyRequestsBufferMode(frags);
877 20 : break;
878 10 : case RequestMode::SequenceID:
879 10 : applyRequestsSequenceIDMode(frags);
880 10 : break;
881 5 : case RequestMode::Ignored:
882 : default:
883 5 : applyRequestsIgnoredMode(frags);
884 5 : break;
885 : }
886 :
887 87 : getDataBuffersStats();
888 :
889 87 : auto stale_requests = requestBuffer_->GetRequests(std::chrono::steady_clock::now() - std::chrono::microseconds(missing_request_window_timeout_us_));
890 35186 : for (auto& req : stale_requests)
891 : {
892 70198 : TLOG(TLVL_CHECKWINDOWS) << "checkSentWindows: Request " << req.first << " has been in the request buffer for more than " << missing_request_window_timeout_us_ << " us, removing from requestBuffer_.";
893 35099 : requestBuffer_->RemoveRequest(req.first);
894 70198 : for (auto& id : dataBuffers_)
895 : {
896 35099 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
897 35099 : if (id.second->WindowsSent.count(req.first))
898 : {
899 0 : id.second->WindowsSent.erase(req.first);
900 : }
901 : else
902 : {
903 70198 : sendEmptyFragment(frags, req.first, id.first, "Request timeout");
904 : }
905 35099 : }
906 : }
907 :
908 87 : if (frags.size() > 0)
909 144 : TLOG(TLVL_APPLYREQUESTS) << "Finished Processing requests, returning " << frags.size() << " fragments";
910 87 : return true;
911 87 : }
912 :
913 35108 : bool artdaq::FragmentBuffer::sendEmptyFragment(artdaq::FragmentPtrs& frags, size_t seqId, Fragment::fragment_id_t fragmentId, std::string desc)
914 : {
915 70216 : TLOG(TLVL_EMPTYFRAGMENT) << desc << " sequence ID " << seqId << ", sending empty fragment";
916 35108 : auto frag = new Fragment();
917 35108 : frag->setSequenceID(seqId);
918 35108 : frag->setFragmentID(fragmentId);
919 35108 : frag->setSystemType(Fragment::EmptyFragmentType);
920 35108 : frags.emplace_back(FragmentPtr(frag));
921 35108 : return true;
922 : }
923 :
924 264947 : void artdaq::FragmentBuffer::checkSentWindows(artdaq::Fragment::sequence_id_t seq)
925 : {
926 529894 : TLOG(TLVL_CHECKWINDOWS) << "checkSentWindows: Checking if request " << seq << " can be removed from request list";
927 264947 : bool seqComplete = true;
928 529926 : for (auto& id : dataBuffers_)
929 : {
930 264979 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
931 264979 : if (!id.second->WindowsSent.count(seq))
932 : {
933 21 : seqComplete = false;
934 : }
935 264979 : }
936 264947 : if (seqComplete)
937 : {
938 529864 : TLOG(TLVL_CHECKWINDOWS) << "checkSentWindows: Request " << seq << " is complete, removing from requestBuffer_.";
939 264932 : requestBuffer_->RemoveRequest(seq);
940 :
941 529886 : for (auto& id : dataBuffers_)
942 : {
943 264954 : std::lock_guard<std::mutex> lk(id.second->DataBufferMutex);
944 264954 : id.second->WindowsSent.erase(seq);
945 264954 : }
946 : }
947 264947 : }
|
