Line data Source code
1 : #define TRACE_NAME "FragmentBuffer_t"
2 :
3 : #define BOOST_TEST_MODULE FragmentBuffer_t
4 : #include <boost/test/unit_test.hpp>
5 :
6 : #include "artdaq-core/Data/ContainerFragment.hh"
7 : #include "artdaq-core/Data/Fragment.hh"
8 : #include "artdaq-core/Utilities/configureMessageFacility.hh"
9 : #include "artdaq/DAQrate/FragmentBuffer.hh"
10 : #include "artdaq/DAQrate/detail/RequestSender.hh"
11 :
12 : #include <thread>
13 :
14 : #define MESSAGEFACILITY_DEBUG true
15 :
16 : #define RATE_TEST_COUNT 100000
17 : #define TRACE_REQUIRE_EQUAL(l, r) \
18 : do \
19 : { \
20 : if (l == r) \
21 : { \
22 : TLOG(TLVL_DEBUG) << __LINE__ << ": Checking if " << #l << " (" << l << ") equals " << #r << " (" << r << ")...YES!"; \
23 : } \
24 : else \
25 : { \
26 : TLOG(TLVL_ERROR) << __LINE__ << ": Checking if " << #l << " (" << l << ") equals " << #r << " (" << r << ")...NO!"; \
27 : } \
28 : BOOST_REQUIRE_EQUAL(l, r); \
29 : } while (0)
30 :
31 : namespace artdaqtest {
32 : class FragmentBufferTestGenerator;
33 : }
34 :
35 : /**
36 : * \brief CommandableFragmentGenerator derived class for testing
37 : */
38 : class artdaqtest::FragmentBufferTestGenerator
39 : {
40 : public:
41 : /**
42 : * \brief FragmentBufferTestGenerator Constructor
43 : */
44 : explicit FragmentBufferTestGenerator(const fhicl::ParameterSet& ps);
45 :
46 : /**
47 : * @brief Generate Fragments
48 : * @param n Number of Fragments to generate
49 : * @param fragmentIds List of Fragment IDs to generate Fragments for (if different than configured fragment IDs)
50 : * @return artdaq::FragmentPtrs containing generated Fragments
51 : */
52 : artdaq::FragmentPtrs Generate(size_t n, std::vector<artdaq::Fragment::fragment_id_t> fragmentIds = std::vector<artdaq::Fragment::fragment_id_t>());
53 :
54 : public:
55 : /**
56 : * \brief Set the timestamp to be used for the next Fragment
57 : * \param ts Timestamp to be used for the next Fragment
58 : */
59 1 : void setTimestamp(artdaq::Fragment::timestamp_t ts) { ts_ = ts; }
60 :
61 : /**
62 : * \brief Get the timestamp that will be used for the next Fragment
63 : * \return The timestamp that will be used for the next Fragment
64 : */
65 : artdaq::Fragment::timestamp_t getTimestamp() { return ts_; }
66 :
67 : private:
68 : artdaq::Fragment::timestamp_t ts_;
69 : artdaq::Fragment::sequence_id_t seq_;
70 : std::set<artdaq::Fragment::fragment_id_t> fragmentIDs_;
71 : };
72 :
73 25 : artdaqtest::FragmentBufferTestGenerator::FragmentBufferTestGenerator(const fhicl::ParameterSet& ps)
74 25 : : ts_(0), seq_(1)
75 : {
76 100 : metricMan->initialize(ps.get<fhicl::ParameterSet>("metrics", fhicl::ParameterSet()));
77 25 : metricMan->do_start();
78 :
79 150 : auto idlist = ps.get<std::vector<artdaq::Fragment::fragment_id_t>>("fragment_ids", {ps.get<artdaq::Fragment::fragment_id_t>("fragment_id", 1)});
80 62 : for (auto& id : idlist)
81 : {
82 37 : fragmentIDs_.insert(id);
83 : }
84 25 : }
85 :
86 300054 : artdaq::FragmentPtrs artdaqtest::FragmentBufferTestGenerator::Generate(size_t n, std::vector<artdaq::Fragment::fragment_id_t> fragmentIds)
87 : {
88 300054 : if (fragmentIds.size() == 0) std::copy(fragmentIDs_.begin(), fragmentIDs_.end(), std::back_inserter(fragmentIds));
89 :
90 300054 : artdaq::FragmentPtrs frags;
91 800202 : while (n > 0)
92 : {
93 500148 : ++ts_;
94 1000371 : for (auto& id : fragmentIds)
95 : {
96 1500669 : TLOG(TLVL_DEBUG) << "Adding Fragment with ID " << id << ", SeqID " << seq_ << ", and timestamp " << ts_;
97 500223 : frags.emplace_back(new artdaq::Fragment(seq_, id, artdaq::Fragment::FirstUserFragmentType, ts_));
98 : }
99 500148 : ++seq_;
100 500148 : n--;
101 : }
102 :
103 300054 : return frags;
104 0 : }
105 :
106 : BOOST_AUTO_TEST_SUITE(FragmentBuffer_t)
107 :
108 2 : BOOST_AUTO_TEST_CASE(ImproperConfiguration)
109 : {
110 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
111 3 : TLOG(TLVL_INFO) << "ImproperConfiguration test case BEGIN";
112 1 : fhicl::ParameterSet ps;
113 2 : ps.put<int>("fragment_id", 1);
114 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
115 3 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
116 3 : ps.put<std::string>("request_mode", "window");
117 :
118 1 : artdaq::FragmentBuffer fp(ps);
119 1 : BOOST_REQUIRE_EQUAL(static_cast<int>(fp.request_mode()), static_cast<int>(artdaq::RequestMode::Ignored));
120 :
121 2 : ps.put<bool>("receive_requests", true);
122 1 : artdaq::FragmentBuffer fpp(ps);
123 1 : BOOST_REQUIRE_EQUAL(static_cast<int>(fpp.request_mode()), static_cast<int>(artdaq::RequestMode::Window));
124 :
125 5 : ps.put<std::vector<int>>("fragment_ids", {2, 3, 4});
126 3 : BOOST_REQUIRE_EXCEPTION(artdaq::FragmentBuffer ffp(ps), cet::exception, [](cet::exception const& e) { return e.category() == "FragmentBufferConfig"; });
127 :
128 3 : TLOG(TLVL_INFO) << "ImproperConfiguration test case END";
129 1 : }
130 :
131 2 : BOOST_AUTO_TEST_CASE(IgnoreRequests)
132 : {
133 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
134 3 : TLOG(TLVL_INFO) << "IgnoreRequests test case BEGIN";
135 1 : fhicl::ParameterSet ps;
136 2 : ps.put<int>("fragment_id", 1);
137 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
138 3 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
139 3 : ps.put<std::string>("request_mode", "ignored");
140 :
141 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
142 1 : buffer->setRunning(true);
143 :
144 1 : artdaq::FragmentBuffer fp(ps);
145 1 : fp.SetRequestBuffer(buffer);
146 :
147 1 : buffer->push(53, 35);
148 :
149 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
150 :
151 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
152 :
153 1 : artdaq::FragmentPtrs fps;
154 1 : auto sts = fp.applyRequests(fps);
155 :
156 4 : TRACE_REQUIRE_EQUAL(sts, true);
157 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
158 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
159 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
160 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
161 :
162 3 : TLOG(TLVL_INFO) << "IgnoreRequests test case END";
163 1 : }
164 :
165 2 : BOOST_AUTO_TEST_CASE(SingleMode)
166 : {
167 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
168 3 : TLOG(TLVL_INFO) << "SingleMode test case BEGIN";
169 1 : fhicl::ParameterSet ps;
170 2 : ps.put<int>("fragment_id", 1);
171 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
172 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
173 3 : ps.put<bool>("receive_requests", true);
174 3 : ps.put<std::string>("request_mode", "single");
175 :
176 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
177 1 : buffer->setRunning(true);
178 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
179 1 : artdaq::FragmentBuffer fp(ps);
180 1 : fp.SetRequestBuffer(buffer);
181 :
182 1 : buffer->push(1, 1);
183 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
184 :
185 1 : artdaq::FragmentPtrs fps;
186 1 : auto sts = fp.applyRequests(fps);
187 1 : auto type = artdaq::Fragment::FirstUserFragmentType;
188 4 : TRACE_REQUIRE_EQUAL(sts, true);
189 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
190 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
191 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
192 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
193 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
194 1 : fps.clear();
195 :
196 1 : buffer->push(2, 5);
197 1 : sts = fp.applyRequests(fps);
198 4 : TRACE_REQUIRE_EQUAL(sts, true);
199 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
200 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
201 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
202 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
203 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
204 1 : fps.clear();
205 :
206 : // Out of order request
207 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
208 1 : buffer->push(4, 7);
209 :
210 1 : sts = fp.applyRequests(fps);
211 4 : TRACE_REQUIRE_EQUAL(sts, true);
212 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
213 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
214 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 7);
215 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
216 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
217 1 : fps.clear();
218 :
219 : // Multiple requests between processing steps will result in an empty
220 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
221 1 : buffer->push(5, 8);
222 1 : buffer->push(6, 9);
223 :
224 1 : sts = fp.applyRequests(fps);
225 4 : TRACE_REQUIRE_EQUAL(sts, true);
226 4 : TRACE_REQUIRE_EQUAL(fps.size(), 2);
227 :
228 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
229 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), artdaq::Fragment::InvalidTimestamp);
230 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 5);
231 1 : auto emtpy_type = artdaq::Fragment::EmptyFragmentType;
232 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), emtpy_type);
233 1 : fps.pop_front();
234 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
235 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 9);
236 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 6);
237 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
238 1 : fps.clear();
239 :
240 3 : TLOG(TLVL_INFO) << "SingleMode test case END";
241 1 : }
242 :
243 2 : BOOST_AUTO_TEST_CASE(BufferMode)
244 : {
245 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
246 3 : TLOG(TLVL_INFO) << "BufferMode test case BEGIN";
247 1 : fhicl::ParameterSet ps;
248 2 : ps.put<int>("fragment_id", 1);
249 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
250 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
251 3 : ps.put<bool>("receive_requests", true);
252 3 : ps.put<std::string>("request_mode", "buffer");
253 :
254 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
255 1 : buffer->setRunning(true);
256 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
257 1 : artdaq::FragmentBuffer fp(ps);
258 1 : fp.SetRequestBuffer(buffer);
259 :
260 1 : buffer->push(1, 1);
261 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
262 :
263 1 : artdaq::FragmentPtrs fps;
264 1 : auto sts = fp.applyRequests(fps);
265 4 : TRACE_REQUIRE_EQUAL(sts, true);
266 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
267 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
268 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
269 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
270 1 : auto type = artdaq::Fragment::ContainerFragmentType;
271 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
272 1 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
273 1 : auto cf = artdaq::ContainerFragment(*fps.front());
274 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
275 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
276 1 : type = artdaq::Fragment::FirstUserFragmentType;
277 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
278 1 : fps.clear();
279 :
280 1 : buffer->push(2, 5);
281 1 : sts = fp.applyRequests(fps);
282 4 : TRACE_REQUIRE_EQUAL(sts, true);
283 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
284 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
285 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
286 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
287 1 : type = artdaq::Fragment::ContainerFragmentType;
288 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
289 1 : auto cf2 = artdaq::ContainerFragment(*fps.front());
290 4 : TRACE_REQUIRE_EQUAL(cf2.block_count(), 0);
291 4 : TRACE_REQUIRE_EQUAL(cf2.missing_data(), false);
292 1 : type = artdaq::Fragment::EmptyFragmentType;
293 4 : TRACE_REQUIRE_EQUAL(cf2.fragment_type(), type);
294 1 : fps.clear();
295 :
296 : // Request order does not matter, and multiple requests between processing steps will result in only the latest request being fulfilled, but with all available data
297 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
298 1 : buffer->push(4, 7);
299 :
300 1 : sts = fp.applyRequests(fps);
301 4 : TRACE_REQUIRE_EQUAL(sts, true);
302 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
303 :
304 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
305 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 7);
306 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
307 1 : type = artdaq::Fragment::ContainerFragmentType;
308 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
309 1 : auto cf3 = artdaq::ContainerFragment(*fps.front());
310 4 : TRACE_REQUIRE_EQUAL(cf3.block_count(), 2);
311 4 : TRACE_REQUIRE_EQUAL(cf3.missing_data(), false);
312 1 : type = artdaq::Fragment::FirstUserFragmentType;
313 4 : TRACE_REQUIRE_EQUAL(cf3.fragment_type(), type);
314 1 : fps.clear();
315 :
316 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
317 1 : buffer->push(5, 8);
318 1 : buffer->push(6, 10);
319 :
320 1 : sts = fp.applyRequests(fps);
321 4 : TRACE_REQUIRE_EQUAL(sts, true);
322 4 : TRACE_REQUIRE_EQUAL(fps.size(), 2);
323 :
324 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
325 1 : auto ts = artdaq::Fragment::InvalidTimestamp;
326 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), ts);
327 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 5);
328 1 : auto emptyType = artdaq::Fragment::EmptyFragmentType;
329 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), emptyType);
330 4 : TRACE_REQUIRE_EQUAL(fps.front()->size(), artdaq::detail::RawFragmentHeader::num_words());
331 1 : fps.pop_front();
332 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
333 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 10);
334 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 6);
335 1 : type = artdaq::Fragment::ContainerFragmentType;
336 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
337 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
338 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 2);
339 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), false);
340 1 : type = artdaq::Fragment::FirstUserFragmentType;
341 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
342 1 : fps.clear();
343 :
344 3 : TLOG(TLVL_INFO) << "BufferMode test case END";
345 1 : }
346 :
347 2 : BOOST_AUTO_TEST_CASE(BufferMode_KeepLatest)
348 : {
349 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
350 3 : TLOG(TLVL_INFO) << "BufferMode_KeepLatest test case BEGIN";
351 1 : fhicl::ParameterSet ps;
352 2 : ps.put<int>("fragment_id", 1);
353 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
354 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
355 3 : ps.put<bool>("receive_requests", true);
356 3 : ps.put<std::string>("request_mode", "buffer");
357 2 : ps.put("buffer_mode_keep_latest", true);
358 :
359 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
360 1 : buffer->setRunning(true);
361 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
362 1 : artdaq::FragmentBuffer fp(ps);
363 1 : fp.SetRequestBuffer(buffer);
364 :
365 1 : buffer->push(1, 1);
366 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
367 :
368 1 : artdaq::FragmentPtrs fps;
369 1 : auto sts = fp.applyRequests(fps);
370 4 : TRACE_REQUIRE_EQUAL(sts, true);
371 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
372 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
373 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
374 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
375 1 : auto type = artdaq::Fragment::ContainerFragmentType;
376 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
377 1 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
378 1 : auto cf = artdaq::ContainerFragment(*fps.front());
379 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
380 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
381 1 : type = artdaq::Fragment::FirstUserFragmentType;
382 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
383 1 : fps.clear();
384 :
385 1 : buffer->push(2, 5);
386 1 : sts = fp.applyRequests(fps);
387 4 : TRACE_REQUIRE_EQUAL(sts, true);
388 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
389 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
390 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
391 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
392 1 : type = artdaq::Fragment::ContainerFragmentType;
393 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
394 1 : auto cf2 = artdaq::ContainerFragment(*fps.front());
395 4 : TRACE_REQUIRE_EQUAL(cf2.block_count(), 1);
396 4 : TRACE_REQUIRE_EQUAL(cf2.missing_data(), false);
397 1 : type = artdaq::Fragment::FirstUserFragmentType;
398 4 : TRACE_REQUIRE_EQUAL(cf2.fragment_type(), type);
399 1 : fps.clear();
400 :
401 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
402 1 : buffer->push(4, 7);
403 1 : sts = fp.applyRequests(fps);
404 4 : TRACE_REQUIRE_EQUAL(sts, true);
405 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
406 :
407 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
408 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 7);
409 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
410 1 : type = artdaq::Fragment::ContainerFragmentType;
411 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
412 1 : auto cf3 = artdaq::ContainerFragment(*fps.front());
413 4 : TRACE_REQUIRE_EQUAL(cf3.block_count(), 2);
414 4 : TRACE_REQUIRE_EQUAL(cf3.missing_data(), false);
415 1 : type = artdaq::Fragment::FirstUserFragmentType;
416 4 : TRACE_REQUIRE_EQUAL(cf3.fragment_type(), type);
417 1 : fps.clear();
418 :
419 1 : buffer->push(5, 8);
420 1 : buffer->push(6, 10);
421 1 : sts = fp.applyRequests(fps);
422 4 : TRACE_REQUIRE_EQUAL(sts, true);
423 4 : TRACE_REQUIRE_EQUAL(fps.size(), 2);
424 :
425 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
426 1 : auto ts = artdaq::Fragment::InvalidTimestamp;
427 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), ts);
428 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 5);
429 1 : auto emptyType = artdaq::Fragment::EmptyFragmentType;
430 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), emptyType);
431 4 : TRACE_REQUIRE_EQUAL(fps.front()->size(), artdaq::detail::RawFragmentHeader::num_words());
432 1 : fps.pop_front();
433 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
434 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 10);
435 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 6);
436 1 : type = artdaq::Fragment::ContainerFragmentType;
437 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
438 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
439 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 1);
440 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), false);
441 1 : type = artdaq::Fragment::FirstUserFragmentType;
442 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
443 1 : fps.clear();
444 :
445 3 : TLOG(TLVL_INFO) << "BufferMode_KeepLatest test case END";
446 1 : }
447 2 : BOOST_AUTO_TEST_CASE(CircularBufferMode)
448 : {
449 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
450 3 : TLOG(TLVL_INFO) << "CircularBufferMode test case BEGIN";
451 1 : fhicl::ParameterSet ps;
452 2 : ps.put<int>("fragment_id", 1);
453 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
454 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
455 2 : ps.put<bool>("receive_requests", true);
456 2 : ps.put<bool>("circular_buffer_mode", true);
457 3 : ps.put<int>("data_buffer_depth_fragments", 3);
458 3 : ps.put<std::string>("request_mode", "buffer");
459 :
460 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
461 1 : buffer->setRunning(true);
462 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
463 1 : artdaq::FragmentBuffer fp(ps);
464 1 : fp.SetRequestBuffer(buffer);
465 :
466 1 : buffer->push(1, 1);
467 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
468 :
469 1 : artdaq::FragmentPtrs fps;
470 1 : auto sts = fp.applyRequests(fps);
471 4 : TRACE_REQUIRE_EQUAL(sts, true);
472 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
473 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
474 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
475 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
476 1 : auto type = artdaq::Fragment::ContainerFragmentType;
477 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
478 1 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
479 1 : auto cf = artdaq::ContainerFragment(*fps.front());
480 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
481 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
482 1 : type = artdaq::Fragment::FirstUserFragmentType;
483 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
484 1 : fps.clear();
485 :
486 1 : buffer->push(2, 5);
487 1 : sts = fp.applyRequests(fps);
488 4 : TRACE_REQUIRE_EQUAL(sts, true);
489 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
490 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
491 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
492 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
493 1 : type = artdaq::Fragment::ContainerFragmentType;
494 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
495 1 : auto cf2 = artdaq::ContainerFragment(*fps.front());
496 4 : TRACE_REQUIRE_EQUAL(cf2.block_count(), 0);
497 4 : TRACE_REQUIRE_EQUAL(cf2.missing_data(), false);
498 1 : type = artdaq::Fragment::EmptyFragmentType;
499 4 : TRACE_REQUIRE_EQUAL(cf2.fragment_type(), type);
500 1 : fps.clear();
501 :
502 1 : fp.AddFragmentsToBuffer(gen.Generate(3));
503 :
504 1 : buffer->push(4, 7);
505 1 : sts = fp.applyRequests(fps);
506 4 : TRACE_REQUIRE_EQUAL(sts, true);
507 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
508 :
509 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
510 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 7);
511 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
512 1 : type = artdaq::Fragment::ContainerFragmentType;
513 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
514 1 : auto cf3 = artdaq::ContainerFragment(*fps.front());
515 4 : TRACE_REQUIRE_EQUAL(cf3.block_count(), 3);
516 4 : TRACE_REQUIRE_EQUAL(cf3.missing_data(), false);
517 1 : type = artdaq::Fragment::FirstUserFragmentType;
518 4 : TRACE_REQUIRE_EQUAL(cf3.fragment_type(), type);
519 1 : fps.clear();
520 :
521 1 : fp.AddFragmentsToBuffer(gen.Generate(5));
522 :
523 1 : buffer->push(5, 8);
524 :
525 1 : sts = fp.applyRequests(fps);
526 4 : TRACE_REQUIRE_EQUAL(sts, true);
527 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
528 :
529 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
530 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 8);
531 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 5);
532 1 : type = artdaq::Fragment::ContainerFragmentType;
533 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
534 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
535 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 3);
536 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), false);
537 1 : type = artdaq::Fragment::FirstUserFragmentType;
538 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
539 4 : TRACE_REQUIRE_EQUAL(cf4.at(0)->timestamp(), 7);
540 4 : TRACE_REQUIRE_EQUAL(cf4.at(1)->timestamp(), 8);
541 4 : TRACE_REQUIRE_EQUAL(cf4.at(2)->timestamp(), 9);
542 1 : fps.clear();
543 :
544 3 : TLOG(TLVL_INFO) << "CircularBufferMode test case END";
545 1 : }
546 :
547 2 : BOOST_AUTO_TEST_CASE(WindowMode_Function)
548 : {
549 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
550 3 : TLOG(TLVL_INFO) << "WindowMode_Function test case BEGIN";
551 1 : fhicl::ParameterSet ps;
552 2 : ps.put<int>("fragment_id", 1);
553 :
554 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
555 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
556 2 : ps.put<size_t>("data_buffer_depth_fragments", 5);
557 3 : ps.put<bool>("circular_buffer_mode", true);
558 3 : ps.put<std::string>("request_mode", "window");
559 2 : ps.put<bool>("receive_requests", true);
560 2 : ps.put<size_t>("missing_request_window_timeout_us", 500000);
561 2 : ps.put<size_t>("window_close_timeout_us", 500000);
562 :
563 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
564 1 : buffer->setRunning(true);
565 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
566 1 : artdaq::FragmentBuffer fp(ps);
567 1 : fp.SetRequestBuffer(buffer);
568 :
569 1 : buffer->push(1, 1);
570 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
571 :
572 1 : artdaq::FragmentPtrs fps;
573 1 : auto sts = fp.applyRequests(fps);
574 4 : TRACE_REQUIRE_EQUAL(sts, true);
575 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
576 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
577 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
578 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
579 1 : auto type = artdaq::Fragment::ContainerFragmentType;
580 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
581 1 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
582 1 : auto cf = artdaq::ContainerFragment(*fps.front());
583 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
584 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
585 1 : type = artdaq::Fragment::FirstUserFragmentType;
586 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
587 1 : fps.clear();
588 :
589 : // No data for request
590 1 : buffer->push(2, 2);
591 :
592 1 : sts = fp.applyRequests(fps);
593 4 : TRACE_REQUIRE_EQUAL(sts, true);
594 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
595 :
596 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
597 :
598 1 : sts = fp.applyRequests(fps);
599 4 : TRACE_REQUIRE_EQUAL(sts, true);
600 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
601 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
602 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 2);
603 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
604 1 : type = artdaq::Fragment::ContainerFragmentType;
605 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
606 1 : auto cf2 = artdaq::ContainerFragment(*fps.front());
607 4 : TRACE_REQUIRE_EQUAL(cf2.block_count(), 1);
608 4 : TRACE_REQUIRE_EQUAL(cf2.missing_data(), false);
609 1 : type = artdaq::Fragment::FirstUserFragmentType;
610 4 : TRACE_REQUIRE_EQUAL(cf2.fragment_type(), type);
611 1 : fps.clear();
612 :
613 : // Request Timeout
614 1 : buffer->push(4, 3);
615 :
616 1 : sts = fp.applyRequests(fps);
617 4 : TRACE_REQUIRE_EQUAL(sts, true);
618 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
619 :
620 1 : usleep(1500000);
621 1 : sts = fp.applyRequests(fps);
622 4 : TRACE_REQUIRE_EQUAL(sts, true);
623 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
624 :
625 : // Also, missing request timeout
626 1 : auto list = fp.GetSentWindowList(1);
627 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
628 :
629 1 : usleep(1500000);
630 :
631 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
632 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 3);
633 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
634 1 : type = artdaq::Fragment::ContainerFragmentType;
635 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
636 1 : auto cf3 = artdaq::ContainerFragment(*fps.front());
637 4 : TRACE_REQUIRE_EQUAL(cf3.block_count(), 0);
638 4 : TRACE_REQUIRE_EQUAL(cf3.missing_data(), true);
639 1 : type = artdaq::Fragment::EmptyFragmentType;
640 4 : TRACE_REQUIRE_EQUAL(cf3.fragment_type(), type);
641 1 : fps.clear();
642 :
643 : // Data-taking has passed request
644 1 : fp.AddFragmentsToBuffer(gen.Generate(12));
645 :
646 1 : buffer->push(5, 4);
647 :
648 1 : list = fp.GetSentWindowList(1); // Out-of-order list is only updated in getNext calls
649 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
650 1 : sts = fp.applyRequests(fps);
651 1 : list = fp.GetSentWindowList(1);
652 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
653 4 : TRACE_REQUIRE_EQUAL(sts, true);
654 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
655 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
656 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 4);
657 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 5);
658 1 : type = artdaq::Fragment::ContainerFragmentType;
659 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
660 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
661 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 0);
662 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), true);
663 1 : type = artdaq::Fragment::EmptyFragmentType;
664 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
665 1 : fps.clear();
666 :
667 : // Out-of-order windows
668 1 : buffer->push(7, 13);
669 :
670 1 : sts = fp.applyRequests(fps);
671 4 : TRACE_REQUIRE_EQUAL(sts, true);
672 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
673 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
674 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 13);
675 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 7);
676 1 : type = artdaq::Fragment::ContainerFragmentType;
677 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
678 1 : auto cf5 = artdaq::ContainerFragment(*fps.front());
679 4 : TRACE_REQUIRE_EQUAL(cf5.block_count(), 1);
680 4 : TRACE_REQUIRE_EQUAL(cf5.missing_data(), false);
681 1 : type = artdaq::Fragment::FirstUserFragmentType;
682 4 : TRACE_REQUIRE_EQUAL(cf5.fragment_type(), type);
683 1 : fps.clear();
684 :
685 1 : buffer->push(6, 12);
686 :
687 1 : sts = fp.applyRequests(fps);
688 4 : TRACE_REQUIRE_EQUAL(sts, true);
689 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
690 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
691 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 12);
692 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 6);
693 1 : type = artdaq::Fragment::ContainerFragmentType;
694 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
695 1 : auto cf6 = artdaq::ContainerFragment(*fps.front());
696 4 : TRACE_REQUIRE_EQUAL(cf6.block_count(), 1);
697 4 : TRACE_REQUIRE_EQUAL(cf6.missing_data(), false);
698 1 : type = artdaq::Fragment::FirstUserFragmentType;
699 4 : TRACE_REQUIRE_EQUAL(cf6.fragment_type(), type);
700 1 : fps.clear();
701 :
702 1 : list = fp.GetSentWindowList(1);
703 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
704 :
705 1 : usleep(1500000);
706 :
707 3 : TLOG(TLVL_INFO) << "WindowMode_Function test case END";
708 1 : }
709 :
710 : // 1. Both start and end before any data in buffer "RequestBeforeBuffer"
711 : // 2. Start before buffer, end in buffer "RequestStartsBeforeBuffer"
712 : // 3. Start befoer buffer, end after buffer "RequestOutsideBuffer"
713 : // 4. Start and end in buffer "RequestInBuffer"
714 : // 5. Start in buffer, end after buffer "RequestEndsAfterBuffer"
715 : // 6. Start and end after buffer "RequestAfterBuffer"
716 2 : BOOST_AUTO_TEST_CASE(WindowMode_RequestBeforeBuffer)
717 : {
718 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
719 3 : TLOG(TLVL_INFO) << "WindowMode_RequestBeforeBuffer test case BEGIN";
720 1 : fhicl::ParameterSet ps;
721 2 : ps.put<int>("fragment_id", 1);
722 :
723 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
724 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 3);
725 2 : ps.put<bool>("circular_buffer_mode", true);
726 2 : ps.put<bool>("receive_requests", true);
727 3 : ps.put<size_t>("data_buffer_depth_fragments", 5);
728 3 : ps.put<std::string>("request_mode", "window");
729 :
730 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
731 1 : buffer->setRunning(true);
732 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
733 1 : artdaq::FragmentBuffer fp(ps);
734 1 : fp.SetRequestBuffer(buffer);
735 :
736 1 : artdaq::FragmentPtrs fps;
737 : int sts;
738 : artdaq::Fragment::type_t type;
739 :
740 : // 1. Both start and end before any data in buffer
741 : // -- Should return ContainerFragment with MissingData bit set and zero Fragments
742 1 : fp.AddFragmentsToBuffer(gen.Generate(10)); // Buffer start is at ts 6, end at 10
743 :
744 1 : buffer->push(1, 1); // Requesting data from ts 1 to 3
745 :
746 1 : sts = fp.applyRequests(fps);
747 4 : TRACE_REQUIRE_EQUAL(sts, true);
748 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
749 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
750 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
751 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
752 1 : type = artdaq::Fragment::ContainerFragmentType;
753 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
754 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
755 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 0);
756 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), true);
757 1 : type = artdaq::Fragment::EmptyFragmentType;
758 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
759 :
760 3 : TLOG(TLVL_INFO) << "WindowMode_RequestBeforeBuffer test case END";
761 1 : }
762 2 : BOOST_AUTO_TEST_CASE(WindowMode_RequestStartsBeforeBuffer)
763 : {
764 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
765 3 : TLOG(TLVL_INFO) << "WindowMode_RequestStartsBeforeBuffer test case BEGIN";
766 :
767 1 : fhicl::ParameterSet ps;
768 2 : ps.put<int>("fragment_id", 1);
769 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
770 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 3);
771 2 : ps.put<size_t>("data_buffer_depth_fragments", 5);
772 2 : ps.put<bool>("circular_buffer_mode", true);
773 3 : ps.put<bool>("receive_requests", true);
774 3 : ps.put<std::string>("request_mode", "window");
775 :
776 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
777 1 : buffer->setRunning(true);
778 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
779 1 : artdaq::FragmentBuffer fp(ps);
780 1 : fp.SetRequestBuffer(buffer);
781 :
782 1 : artdaq::FragmentPtrs fps;
783 : int sts;
784 : artdaq::Fragment::type_t type;
785 :
786 1 : fp.AddFragmentsToBuffer(gen.Generate(10)); // Buffer contains 6 to 10
787 :
788 : // 2. Start before buffer, end in buffer
789 : // -- Should return ContainerFragment with MissingData bit set and one or more Fragments
790 1 : buffer->push(1, 4); // Requesting data from ts 4 to 6
791 :
792 1 : sts = fp.applyRequests(fps);
793 4 : TRACE_REQUIRE_EQUAL(sts, true);
794 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
795 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
796 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 4);
797 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
798 1 : type = artdaq::Fragment::ContainerFragmentType;
799 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
800 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
801 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 1);
802 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), true);
803 1 : type = artdaq::Fragment::FirstUserFragmentType;
804 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
805 :
806 3 : TLOG(TLVL_INFO) << "WindowMode_RequestStartsBeforeBuffer test case END";
807 1 : }
808 2 : BOOST_AUTO_TEST_CASE(WindowMode_RequestOutsideBuffer)
809 : {
810 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
811 3 : TLOG(TLVL_INFO) << "WindowMode_RequestOutsideBuffer test case BEGIN";
812 1 : fhicl::ParameterSet ps;
813 2 : ps.put<int>("fragment_id", 1);
814 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
815 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 4);
816 2 : ps.put<size_t>("window_close_timeout_us", 500000);
817 2 : ps.put<bool>("circular_buffer_mode", true);
818 2 : ps.put<size_t>("data_buffer_depth_fragments", 5);
819 3 : ps.put<bool>("receive_requests", true);
820 3 : ps.put<std::string>("request_mode", "window");
821 :
822 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
823 1 : buffer->setRunning(true);
824 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
825 1 : artdaq::FragmentBuffer fp(ps);
826 1 : fp.SetRequestBuffer(buffer);
827 :
828 1 : artdaq::FragmentPtrs fps;
829 : int sts;
830 : artdaq::Fragment::type_t type;
831 :
832 1 : fp.AddFragmentsToBuffer(gen.Generate(10)); // Buffer contains 6 to 10
833 :
834 : // 3. Start before buffer, end after buffer
835 : // -- Should not return until buffer passes end or timeout (check both cases), MissingData bit set
836 :
837 1 : buffer->push(1, 6); // Requesting data from ts 6 to 9, buffer will contain 10
838 :
839 1 : sts = fp.applyRequests(fps);
840 4 : TRACE_REQUIRE_EQUAL(sts, true);
841 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
842 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
843 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 6);
844 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
845 1 : type = artdaq::Fragment::ContainerFragmentType;
846 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
847 1 : auto cf = artdaq::ContainerFragment(*fps.front());
848 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 4);
849 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
850 1 : type = artdaq::Fragment::FirstUserFragmentType;
851 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
852 1 : fps.clear();
853 :
854 1 : buffer->push(2, 9); // Requesting data from ts 9 to 12
855 :
856 1 : sts = fp.applyRequests(fps);
857 4 : TRACE_REQUIRE_EQUAL(sts, true);
858 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
859 :
860 1 : fp.AddFragmentsToBuffer(gen.Generate(3)); // Buffer start is at ts 10, end at 13
861 :
862 1 : sts = fp.applyRequests(fps);
863 4 : TRACE_REQUIRE_EQUAL(sts, true);
864 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
865 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
866 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 9);
867 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
868 1 : type = artdaq::Fragment::ContainerFragmentType;
869 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
870 1 : auto cf2 = artdaq::ContainerFragment(*fps.front());
871 4 : TRACE_REQUIRE_EQUAL(cf2.block_count(), 3);
872 4 : TRACE_REQUIRE_EQUAL(cf2.missing_data(), true);
873 1 : type = artdaq::Fragment::FirstUserFragmentType;
874 4 : TRACE_REQUIRE_EQUAL(cf2.fragment_type(), type);
875 1 : fps.clear();
876 :
877 1 : buffer->push(3, 12); // Requesting data from ts 11 to 14
878 :
879 1 : sts = fp.applyRequests(fps);
880 4 : TRACE_REQUIRE_EQUAL(sts, true);
881 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
882 :
883 1 : usleep(550000);
884 :
885 1 : sts = fp.applyRequests(fps);
886 4 : TRACE_REQUIRE_EQUAL(sts, true);
887 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
888 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
889 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 12);
890 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 3);
891 1 : type = artdaq::Fragment::ContainerFragmentType;
892 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
893 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
894 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 1);
895 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), true);
896 1 : type = artdaq::Fragment::FirstUserFragmentType;
897 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
898 :
899 3 : TLOG(TLVL_INFO) << "WindowMode_RequestOutsideBuffer test case END";
900 1 : }
901 2 : BOOST_AUTO_TEST_CASE(WindowMode_RequestInBuffer)
902 : {
903 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
904 3 : TLOG(TLVL_INFO) << "WindowMode_RequestInBuffer test case BEGIN";
905 1 : fhicl::ParameterSet ps;
906 2 : ps.put<int>("fragment_id", 1);
907 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
908 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 3);
909 2 : ps.put<bool>("circular_buffer_mode", true);
910 2 : ps.put<size_t>("data_buffer_depth_fragments", 5);
911 3 : ps.put<bool>("receive_requests", true);
912 3 : ps.put<std::string>("request_mode", "window");
913 :
914 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
915 1 : buffer->setRunning(true);
916 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
917 1 : artdaq::FragmentBuffer fp(ps);
918 1 : fp.SetRequestBuffer(buffer);
919 :
920 1 : artdaq::FragmentPtrs fps;
921 : int sts;
922 : artdaq::Fragment::type_t type;
923 :
924 : // 4. Start and end in buffer
925 : // -- Should return ContainerFragment with one or more Fragments
926 1 : fp.AddFragmentsToBuffer(gen.Generate(6)); // Buffer start is at ts 2, end at 6
927 :
928 1 : buffer->push(1, 3); // Requesting data from ts 3 to 5
929 :
930 1 : sts = fp.applyRequests(fps);
931 4 : TRACE_REQUIRE_EQUAL(sts, true);
932 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
933 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
934 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 3);
935 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
936 1 : type = artdaq::Fragment::ContainerFragmentType;
937 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
938 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
939 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 3);
940 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), false);
941 1 : type = artdaq::Fragment::FirstUserFragmentType;
942 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
943 :
944 3 : TLOG(TLVL_INFO) << "WindowMode_RequestInBuffer test case END";
945 1 : }
946 2 : BOOST_AUTO_TEST_CASE(WindowMode_RequestEndsAfterBuffer)
947 : {
948 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
949 3 : TLOG(TLVL_INFO) << "WindowMode_RequestEndsAfterBuffer test case BEGIN";
950 1 : fhicl::ParameterSet ps;
951 2 : ps.put<int>("fragment_id", 1);
952 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
953 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 3);
954 2 : ps.put<size_t>("window_close_timeout_us", 500000);
955 2 : ps.put<bool>("circular_buffer_mode", true);
956 2 : ps.put<bool>("receive_requests", true);
957 3 : ps.put<size_t>("data_buffer_depth_fragments", 5);
958 3 : ps.put<std::string>("request_mode", "window");
959 :
960 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
961 1 : buffer->setRunning(true);
962 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
963 1 : artdaq::FragmentBuffer fp(ps);
964 1 : fp.SetRequestBuffer(buffer);
965 :
966 1 : artdaq::FragmentPtrs fps;
967 : int sts;
968 : artdaq::Fragment::type_t type;
969 :
970 1 : fp.AddFragmentsToBuffer(gen.Generate(6)); // Buffer contains 2 to 6
971 :
972 : // 5. Start in buffer, end after buffer
973 : // -- Should not return until buffer passes end or timeout (check both cases). MissingData bit set if timeout
974 1 : buffer->push(1, 5); // Requesting data from ts 5 to 7
975 :
976 1 : sts = fp.applyRequests(fps);
977 4 : TRACE_REQUIRE_EQUAL(sts, true);
978 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
979 :
980 1 : fp.AddFragmentsToBuffer(gen.Generate(2)); // Buffer contains 4 to 8
981 :
982 1 : sts = fp.applyRequests(fps);
983 4 : TRACE_REQUIRE_EQUAL(sts, true);
984 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
985 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
986 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
987 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
988 1 : type = artdaq::Fragment::ContainerFragmentType;
989 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
990 1 : auto cf = artdaq::ContainerFragment(*fps.front());
991 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 3);
992 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
993 1 : type = artdaq::Fragment::FirstUserFragmentType;
994 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
995 1 : fps.clear();
996 :
997 1 : buffer->push(2, 8); // Requesting data from ts 8 to 10
998 :
999 1 : sts = fp.applyRequests(fps);
1000 4 : TRACE_REQUIRE_EQUAL(sts, true);
1001 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
1002 :
1003 1 : usleep(550000);
1004 :
1005 1 : sts = fp.applyRequests(fps);
1006 4 : TRACE_REQUIRE_EQUAL(sts, true);
1007 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
1008 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
1009 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 8);
1010 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1011 1 : type = artdaq::Fragment::ContainerFragmentType;
1012 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1013 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
1014 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 1);
1015 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), true);
1016 1 : type = artdaq::Fragment::FirstUserFragmentType;
1017 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
1018 :
1019 3 : TLOG(TLVL_INFO) << "WindowMode_RequestEndsAfterBuffer test case END";
1020 1 : }
1021 2 : BOOST_AUTO_TEST_CASE(WindowMode_RequestAfterBuffer)
1022 : {
1023 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1024 3 : TLOG(TLVL_INFO) << "WindowMode_RequestAfterBuffer test case BEGIN";
1025 1 : fhicl::ParameterSet ps;
1026 2 : ps.put<int>("fragment_id", 1);
1027 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1028 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 3);
1029 2 : ps.put<size_t>("window_close_timeout_us", 500000);
1030 2 : ps.put<bool>("circular_buffer_mode", true);
1031 2 : ps.put<bool>("receive_requests", true);
1032 3 : ps.put<size_t>("data_buffer_depth_fragments", 5);
1033 3 : ps.put<std::string>("request_mode", "window");
1034 :
1035 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1036 1 : buffer->setRunning(true);
1037 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1038 1 : artdaq::FragmentBuffer fp(ps);
1039 1 : fp.SetRequestBuffer(buffer);
1040 :
1041 1 : artdaq::FragmentPtrs fps;
1042 : int sts;
1043 : artdaq::Fragment::type_t type;
1044 :
1045 : // 6. Start and end after buffer
1046 : // -- Should not return until buffer passes end or timeout (check both cases). MissingData bit set if timeout
1047 1 : fp.AddFragmentsToBuffer(gen.Generate(10)); // Buffer start is 6, end at 10
1048 :
1049 1 : buffer->push(1, 11); // Requesting data from ts 11 to 13
1050 :
1051 1 : sts = fp.applyRequests(fps);
1052 4 : TRACE_REQUIRE_EQUAL(sts, true);
1053 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
1054 1 : fp.AddFragmentsToBuffer(gen.Generate(1)); // Buffer start is 7, end at 11
1055 :
1056 1 : sts = fp.applyRequests(fps);
1057 4 : TRACE_REQUIRE_EQUAL(sts, true);
1058 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
1059 :
1060 1 : fp.AddFragmentsToBuffer(gen.Generate(3)); // Buffer start is 10, end at 14
1061 :
1062 1 : sts = fp.applyRequests(fps);
1063 4 : TRACE_REQUIRE_EQUAL(sts, true);
1064 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
1065 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
1066 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 11);
1067 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1068 1 : type = artdaq::Fragment::ContainerFragmentType;
1069 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1070 1 : auto cf = artdaq::ContainerFragment(*fps.front());
1071 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 3);
1072 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1073 1 : type = artdaq::Fragment::FirstUserFragmentType;
1074 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1075 1 : fps.clear();
1076 :
1077 1 : buffer->push(2, 16); // Requesting data from ts 15 to 17
1078 :
1079 1 : sts = fp.applyRequests(fps);
1080 4 : TRACE_REQUIRE_EQUAL(sts, true);
1081 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
1082 :
1083 1 : usleep(550000);
1084 :
1085 1 : sts = fp.applyRequests(fps);
1086 4 : TRACE_REQUIRE_EQUAL(sts, true);
1087 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
1088 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
1089 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 16);
1090 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1091 1 : type = artdaq::Fragment::ContainerFragmentType;
1092 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1093 1 : auto cf4 = artdaq::ContainerFragment(*fps.front());
1094 4 : TRACE_REQUIRE_EQUAL(cf4.block_count(), 0);
1095 4 : TRACE_REQUIRE_EQUAL(cf4.missing_data(), true);
1096 1 : type = artdaq::Fragment::EmptyFragmentType;
1097 4 : TRACE_REQUIRE_EQUAL(cf4.fragment_type(), type);
1098 :
1099 3 : TLOG(TLVL_INFO) << "WindowMode_RequestAfterBuffer test case END";
1100 1 : }
1101 :
1102 2 : BOOST_AUTO_TEST_CASE(SequenceIDMode)
1103 : {
1104 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1105 3 : TLOG(TLVL_INFO) << "SequenceIDMode test case BEGIN";
1106 1 : fhicl::ParameterSet ps;
1107 2 : ps.put<int>("fragment_id", 1);
1108 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1109 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
1110 3 : ps.put<bool>("receive_requests", true);
1111 3 : ps.put<std::string>("request_mode", "SequenceID");
1112 :
1113 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1114 1 : buffer->setRunning(true);
1115 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1116 1 : artdaq::FragmentBuffer fp(ps);
1117 1 : fp.SetRequestBuffer(buffer);
1118 :
1119 1 : buffer->push(1, 1);
1120 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1121 :
1122 : // Test that Fragment with matching Sequence ID and timestamp is returned
1123 1 : artdaq::FragmentPtrs fps;
1124 1 : auto sts = fp.applyRequests(fps);
1125 1 : auto type = artdaq::Fragment::FirstUserFragmentType;
1126 4 : TRACE_REQUIRE_EQUAL(sts, true);
1127 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
1128 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
1129 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
1130 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1131 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1132 1 : fps.clear();
1133 :
1134 : // Test that no Fragment is returned when one does not exist in the buffer
1135 1 : buffer->push(2, 5);
1136 :
1137 1 : sts = fp.applyRequests(fps);
1138 4 : TRACE_REQUIRE_EQUAL(sts, true);
1139 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0u);
1140 :
1141 : // Test that Fragment with matching Sequence ID and non-matching timestamp is returned
1142 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1143 :
1144 1 : sts = fp.applyRequests(fps);
1145 4 : TRACE_REQUIRE_EQUAL(sts, true);
1146 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
1147 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
1148 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 2);
1149 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1150 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1151 1 : fps.clear();
1152 :
1153 : // Test out-of-order requests, with non-matching timestamps
1154 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
1155 :
1156 1 : buffer->push(4, 7);
1157 :
1158 1 : sts = fp.applyRequests(fps);
1159 4 : TRACE_REQUIRE_EQUAL(sts, true);
1160 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
1161 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
1162 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 4);
1163 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
1164 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1165 1 : fps.clear();
1166 :
1167 1 : buffer->push(3, 6);
1168 :
1169 1 : sts = fp.applyRequests(fps);
1170 4 : TRACE_REQUIRE_EQUAL(sts, true);
1171 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
1172 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
1173 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 3);
1174 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 3);
1175 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1176 :
1177 3 : TLOG(TLVL_INFO) << "SequenceIDMode test case END";
1178 1 : }
1179 :
1180 2 : BOOST_AUTO_TEST_CASE(IgnoreRequests_MultipleIDs)
1181 : {
1182 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1183 3 : TLOG(TLVL_INFO) << "IgnoreRequests_MultipleIDs test case BEGIN";
1184 1 : fhicl::ParameterSet ps;
1185 4 : ps.put<std::vector<int>>("fragment_ids", {1, 2, 3});
1186 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1187 3 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
1188 3 : ps.put<std::string>("request_mode", "ignored");
1189 :
1190 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1191 1 : buffer->setRunning(true);
1192 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1193 1 : artdaq::FragmentBuffer fp(ps);
1194 1 : fp.SetRequestBuffer(buffer);
1195 :
1196 1 : buffer->push(53, 35);
1197 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1198 :
1199 1 : artdaq::FragmentPtrs fps;
1200 1 : std::map<artdaq::Fragment::fragment_id_t, size_t> ids;
1201 1 : auto sts = fp.applyRequests(fps);
1202 4 : TRACE_REQUIRE_EQUAL(sts, true);
1203 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1204 4 : while (fps.size() > 0)
1205 : {
1206 3 : ids[fps.front()->fragmentID()]++;
1207 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
1208 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1209 3 : fps.pop_front();
1210 : }
1211 :
1212 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1213 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1214 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1215 1 : ids.clear();
1216 :
1217 1 : fps.clear();
1218 :
1219 3 : TLOG(TLVL_INFO) << "IgnoreRequests_MultipleIDs test case END";
1220 1 : }
1221 :
1222 2 : BOOST_AUTO_TEST_CASE(SingleMode_MultipleIDs)
1223 : {
1224 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1225 3 : TLOG(TLVL_INFO) << "SingleMode_MultipleIDs test case BEGIN";
1226 1 : fhicl::ParameterSet ps;
1227 4 : ps.put<std::vector<int>>("fragment_ids", {1, 2, 3});
1228 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1229 2 : ps.put<bool>("receive_requests", true);
1230 3 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
1231 3 : ps.put<std::string>("request_mode", "single");
1232 :
1233 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1234 1 : buffer->setRunning(true);
1235 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1236 1 : artdaq::FragmentBuffer fp(ps);
1237 1 : fp.SetRequestBuffer(buffer);
1238 :
1239 1 : buffer->push(1, 1);
1240 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1241 :
1242 1 : artdaq::FragmentPtrs fps;
1243 1 : std::map<artdaq::Fragment::fragment_id_t, size_t> ids;
1244 1 : auto sts = fp.applyRequests(fps);
1245 1 : auto type = artdaq::Fragment::FirstUserFragmentType;
1246 4 : TRACE_REQUIRE_EQUAL(sts, true);
1247 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1248 4 : while (fps.size() > 0)
1249 : {
1250 3 : ids[fps.front()->fragmentID()]++;
1251 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
1252 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1253 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1254 3 : fps.pop_front();
1255 : }
1256 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1257 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1258 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1259 1 : ids.clear();
1260 :
1261 1 : fps.clear();
1262 :
1263 1 : buffer->push(2, 5);
1264 :
1265 1 : sts = fp.applyRequests(fps);
1266 4 : TRACE_REQUIRE_EQUAL(sts, true);
1267 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1268 4 : while (fps.size() > 0)
1269 : {
1270 3 : ids[fps.front()->fragmentID()]++;
1271 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
1272 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1273 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1274 3 : fps.pop_front();
1275 : }
1276 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1277 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1278 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1279 1 : ids.clear();
1280 :
1281 1 : fps.clear();
1282 :
1283 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
1284 :
1285 : // Don't care about skipped requests
1286 1 : buffer->push(4, 7);
1287 :
1288 1 : sts = fp.applyRequests(fps);
1289 4 : TRACE_REQUIRE_EQUAL(sts, true);
1290 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1291 4 : for (auto ii = 0; ii < 3; ++ii)
1292 : {
1293 3 : ids[fps.front()->fragmentID()]++;
1294 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 7);
1295 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
1296 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1297 3 : fps.pop_front();
1298 : }
1299 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1300 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1301 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1302 1 : ids.clear();
1303 1 : fps.clear();
1304 :
1305 1 : buffer->push(5, 8);
1306 1 : buffer->push(6, 10);
1307 :
1308 1 : sts = fp.applyRequests(fps);
1309 4 : TRACE_REQUIRE_EQUAL(sts, true);
1310 4 : TRACE_REQUIRE_EQUAL(fps.size(), 6);
1311 1 : auto ts = artdaq::Fragment::InvalidTimestamp;
1312 1 : auto emptyType = artdaq::Fragment::EmptyFragmentType;
1313 7 : for (auto ii = 0; ii < 6; ++ii)
1314 : {
1315 6 : ids[fps.front()->fragmentID()]++;
1316 6 : if (fps.front()->sequenceID() == 5)
1317 : {
1318 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), ts);
1319 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), emptyType);
1320 : }
1321 3 : else if (fps.front()->sequenceID() == 6)
1322 : {
1323 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 10);
1324 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1325 : }
1326 : else
1327 : {
1328 0 : BOOST_REQUIRE_MESSAGE(false, "Unexpected sequence ID in Fragment");
1329 : }
1330 6 : fps.pop_front();
1331 : }
1332 4 : TRACE_REQUIRE_EQUAL(ids[1], 2);
1333 4 : TRACE_REQUIRE_EQUAL(ids[2], 2);
1334 4 : TRACE_REQUIRE_EQUAL(ids[3], 2);
1335 1 : ids.clear();
1336 1 : fps.clear();
1337 :
1338 : // Single mode should generate 3 Fragments, 2 new ones and one old one
1339 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1340 :
1341 1 : buffer->push(7, 11);
1342 :
1343 1 : sts = fp.applyRequests(fps);
1344 4 : TRACE_REQUIRE_EQUAL(sts, true);
1345 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1346 4 : for (auto ii = 0; ii < 3; ++ii)
1347 : {
1348 3 : ids[fps.front()->fragmentID()]++;
1349 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 11);
1350 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 7);
1351 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1352 3 : fps.pop_front();
1353 : }
1354 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1355 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1356 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1357 1 : ids.clear();
1358 1 : fps.clear();
1359 :
1360 3 : TLOG(TLVL_INFO) << "SingleMode_MultipleIDs test case END";
1361 1 : }
1362 :
1363 2 : BOOST_AUTO_TEST_CASE(BufferMode_MultipleIDs)
1364 : {
1365 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1366 3 : TLOG(TLVL_INFO) << "BufferMode_MultipleIDs test case BEGIN";
1367 1 : fhicl::ParameterSet ps;
1368 4 : ps.put<std::vector<int>>("fragment_ids", {1, 2, 3});
1369 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1370 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
1371 3 : ps.put<bool>("receive_requests", true);
1372 3 : ps.put<std::string>("request_mode", "buffer");
1373 :
1374 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1375 1 : buffer->setRunning(true);
1376 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1377 1 : artdaq::FragmentBuffer fp(ps);
1378 1 : fp.SetRequestBuffer(buffer);
1379 :
1380 1 : buffer->push(1, 1);
1381 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1382 :
1383 1 : artdaq::FragmentPtrs fps;
1384 1 : std::map<artdaq::Fragment::fragment_id_t, size_t> ids;
1385 1 : auto sts = fp.applyRequests(fps);
1386 4 : TRACE_REQUIRE_EQUAL(sts, true);
1387 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1388 1 : auto type = artdaq::Fragment::ContainerFragmentType;
1389 4 : while (fps.size() > 0)
1390 : {
1391 3 : ids[fps.front()->fragmentID()]++;
1392 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
1393 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1394 3 : type = artdaq::Fragment::ContainerFragmentType;
1395 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1396 3 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
1397 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1398 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1399 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1400 3 : type = artdaq::Fragment::FirstUserFragmentType;
1401 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1402 3 : fps.pop_front();
1403 3 : }
1404 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1405 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1406 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1407 1 : ids.clear();
1408 1 : fps.clear();
1409 :
1410 1 : buffer->push(2, 5);
1411 :
1412 1 : sts = fp.applyRequests(fps);
1413 4 : TRACE_REQUIRE_EQUAL(sts, true);
1414 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1415 4 : while (fps.size() > 0)
1416 : {
1417 3 : ids[fps.front()->fragmentID()]++;
1418 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
1419 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1420 3 : type = artdaq::Fragment::ContainerFragmentType;
1421 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1422 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1423 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 0);
1424 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1425 3 : type = artdaq::Fragment::EmptyFragmentType;
1426 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1427 3 : fps.pop_front();
1428 3 : }
1429 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1430 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1431 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1432 1 : ids.clear();
1433 1 : fps.clear();
1434 :
1435 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
1436 :
1437 1 : buffer->push(4, 7);
1438 :
1439 1 : sts = fp.applyRequests(fps);
1440 4 : TRACE_REQUIRE_EQUAL(sts, true);
1441 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1442 :
1443 4 : for (auto ii = 0; ii < 3; ++ii)
1444 : {
1445 3 : ids[fps.front()->fragmentID()]++;
1446 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 7);
1447 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
1448 3 : type = artdaq::Fragment::ContainerFragmentType;
1449 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1450 3 : auto cf3 = artdaq::ContainerFragment(*fps.front());
1451 12 : TRACE_REQUIRE_EQUAL(cf3.block_count(), 2);
1452 12 : TRACE_REQUIRE_EQUAL(cf3.missing_data(), false);
1453 3 : type = artdaq::Fragment::FirstUserFragmentType;
1454 12 : TRACE_REQUIRE_EQUAL(cf3.fragment_type(), type);
1455 3 : fps.pop_front();
1456 3 : }
1457 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1458 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1459 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1460 1 : ids.clear();
1461 :
1462 1 : fps.clear();
1463 :
1464 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
1465 1 : buffer->push(5, 8);
1466 1 : buffer->push(6, 10);
1467 :
1468 1 : sts = fp.applyRequests(fps);
1469 4 : TRACE_REQUIRE_EQUAL(sts, true);
1470 4 : TRACE_REQUIRE_EQUAL(fps.size(), 6);
1471 1 : auto ts = artdaq::Fragment::InvalidTimestamp;
1472 1 : auto emptyType = artdaq::Fragment::EmptyFragmentType;
1473 7 : for (auto ii = 0; ii < 6; ++ii)
1474 : {
1475 6 : ids[fps.front()->fragmentID()]++;
1476 6 : if (fps.front()->sequenceID() == 5)
1477 : {
1478 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), ts);
1479 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), emptyType);
1480 12 : TRACE_REQUIRE_EQUAL(fps.front()->size(), artdaq::detail::RawFragmentHeader::num_words());
1481 : }
1482 3 : else if (fps.front()->sequenceID() == 6)
1483 : {
1484 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 10);
1485 3 : type = artdaq::Fragment::ContainerFragmentType;
1486 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1487 3 : auto cf3 = artdaq::ContainerFragment(*fps.front());
1488 12 : TRACE_REQUIRE_EQUAL(cf3.block_count(), 2);
1489 12 : TRACE_REQUIRE_EQUAL(cf3.missing_data(), false);
1490 3 : type = artdaq::Fragment::FirstUserFragmentType;
1491 12 : TRACE_REQUIRE_EQUAL(cf3.fragment_type(), type);
1492 3 : }
1493 : else
1494 : {
1495 0 : BOOST_REQUIRE_MESSAGE(false, "Unexpected sequence ID in Fragment");
1496 : }
1497 6 : fps.pop_front();
1498 : }
1499 4 : TRACE_REQUIRE_EQUAL(ids[1], 2);
1500 4 : TRACE_REQUIRE_EQUAL(ids[2], 2);
1501 4 : TRACE_REQUIRE_EQUAL(ids[3], 2);
1502 1 : ids.clear();
1503 :
1504 1 : fps.clear();
1505 3 : TLOG(TLVL_INFO) << "BufferMode_MultipleIDs test case END";
1506 1 : }
1507 :
1508 2 : BOOST_AUTO_TEST_CASE(CircularBufferMode_MultipleIDs)
1509 : {
1510 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1511 3 : TLOG(TLVL_INFO) << "CircularBufferMode_MultipleIDs test case BEGIN";
1512 1 : fhicl::ParameterSet ps;
1513 4 : ps.put<std::vector<int>>("fragment_ids", {1, 2, 3});
1514 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1515 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
1516 2 : ps.put<bool>("receive_requests", true);
1517 2 : ps.put<bool>("circular_buffer_mode", true);
1518 3 : ps.put<int>("data_buffer_depth_fragments", 3);
1519 3 : ps.put<std::string>("request_mode", "buffer");
1520 :
1521 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1522 1 : buffer->setRunning(true);
1523 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1524 1 : artdaq::FragmentBuffer fp(ps);
1525 1 : fp.SetRequestBuffer(buffer);
1526 :
1527 1 : buffer->push(1, 1);
1528 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1529 :
1530 1 : artdaq::FragmentPtrs fps;
1531 1 : std::map<artdaq::Fragment::fragment_id_t, size_t> ids;
1532 1 : auto sts = fp.applyRequests(fps);
1533 1 : auto type = artdaq::Fragment::ContainerFragmentType;
1534 4 : TRACE_REQUIRE_EQUAL(sts, true);
1535 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1536 4 : while (fps.size() > 0)
1537 : {
1538 3 : ids[fps.front()->fragmentID()]++;
1539 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
1540 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1541 3 : type = artdaq::Fragment::ContainerFragmentType;
1542 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1543 3 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
1544 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1545 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1546 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1547 3 : type = artdaq::Fragment::FirstUserFragmentType;
1548 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1549 3 : fps.pop_front();
1550 3 : }
1551 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1552 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1553 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1554 1 : ids.clear();
1555 :
1556 1 : fps.clear();
1557 :
1558 1 : buffer->push(2, 5);
1559 :
1560 1 : sts = fp.applyRequests(fps);
1561 4 : TRACE_REQUIRE_EQUAL(sts, true);
1562 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1563 4 : while (fps.size() > 0)
1564 : {
1565 3 : ids[fps.front()->fragmentID()]++;
1566 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
1567 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1568 3 : type = artdaq::Fragment::ContainerFragmentType;
1569 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1570 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1571 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 0);
1572 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1573 3 : type = artdaq::Fragment::EmptyFragmentType;
1574 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1575 3 : fps.pop_front();
1576 3 : }
1577 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1578 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1579 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1580 1 : ids.clear();
1581 :
1582 1 : fps.clear();
1583 :
1584 1 : fp.AddFragmentsToBuffer(gen.Generate(3));
1585 :
1586 1 : buffer->push(4, 7);
1587 :
1588 1 : sts = fp.applyRequests(fps);
1589 4 : TRACE_REQUIRE_EQUAL(sts, true);
1590 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1591 :
1592 4 : for (auto ii = 0; ii < 3; ++ii)
1593 : {
1594 3 : ids[fps.front()->fragmentID()]++;
1595 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 7);
1596 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
1597 3 : type = artdaq::Fragment::ContainerFragmentType;
1598 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1599 3 : auto cf3 = artdaq::ContainerFragment(*fps.front());
1600 12 : TRACE_REQUIRE_EQUAL(cf3.block_count(), 3);
1601 12 : TRACE_REQUIRE_EQUAL(cf3.missing_data(), false);
1602 3 : type = artdaq::Fragment::FirstUserFragmentType;
1603 12 : TRACE_REQUIRE_EQUAL(cf3.fragment_type(), type);
1604 3 : fps.pop_front();
1605 3 : }
1606 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1607 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1608 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1609 1 : ids.clear();
1610 :
1611 1 : fps.clear();
1612 :
1613 1 : fp.AddFragmentsToBuffer(gen.Generate(5));
1614 :
1615 1 : buffer->push(5, 8);
1616 :
1617 1 : sts = fp.applyRequests(fps);
1618 4 : TRACE_REQUIRE_EQUAL(sts, true);
1619 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1620 4 : while (fps.size() > 0)
1621 : {
1622 3 : ids[fps.front()->fragmentID()]++;
1623 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 8);
1624 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 5);
1625 3 : type = artdaq::Fragment::ContainerFragmentType;
1626 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1627 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1628 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 3);
1629 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1630 3 : type = artdaq::Fragment::FirstUserFragmentType;
1631 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1632 12 : TRACE_REQUIRE_EQUAL(cf.at(0)->timestamp(), 7);
1633 12 : TRACE_REQUIRE_EQUAL(cf.at(1)->timestamp(), 8);
1634 12 : TRACE_REQUIRE_EQUAL(cf.at(2)->timestamp(), 9);
1635 3 : fps.pop_front();
1636 3 : }
1637 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1638 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1639 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1640 1 : ids.clear();
1641 :
1642 1 : fps.clear();
1643 :
1644 3 : TLOG(TLVL_INFO) << "CircularBufferMode_MultipleIDs test case END";
1645 1 : }
1646 :
1647 2 : BOOST_AUTO_TEST_CASE(WindowMode_Function_MultipleIDs)
1648 : {
1649 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1650 3 : TLOG(TLVL_INFO) << "WindowMode_Function_MultipleIDs test case BEGIN";
1651 1 : fhicl::ParameterSet ps;
1652 4 : ps.put<std::vector<int>>("fragment_ids", {1, 2, 3});
1653 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1654 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
1655 2 : ps.put<size_t>("data_buffer_depth_fragments", 5);
1656 2 : ps.put<bool>("circular_buffer_mode", true);
1657 3 : ps.put<bool>("receive_requests", true);
1658 3 : ps.put<std::string>("request_mode", "window");
1659 2 : ps.put<size_t>("missing_request_window_timeout_us", 500000);
1660 2 : ps.put<size_t>("window_close_timeout_us", 500000);
1661 :
1662 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1663 1 : buffer->setRunning(true);
1664 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1665 1 : artdaq::FragmentBuffer fp(ps);
1666 1 : fp.SetRequestBuffer(buffer);
1667 :
1668 1 : buffer->push(1, 1);
1669 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1670 :
1671 1 : artdaq::FragmentPtrs fps;
1672 1 : std::map<artdaq::Fragment::fragment_id_t, size_t> ids;
1673 1 : auto sts = fp.applyRequests(fps);
1674 1 : auto type = artdaq::Fragment::ContainerFragmentType;
1675 4 : TRACE_REQUIRE_EQUAL(sts, true);
1676 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1677 4 : while (fps.size() > 0)
1678 : {
1679 3 : ids[fps.front()->fragmentID()]++;
1680 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
1681 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1682 3 : type = artdaq::Fragment::ContainerFragmentType;
1683 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1684 3 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
1685 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1686 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1687 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1688 3 : type = artdaq::Fragment::FirstUserFragmentType;
1689 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1690 3 : fps.pop_front();
1691 3 : }
1692 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1693 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1694 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1695 1 : ids.clear();
1696 1 : fps.clear();
1697 :
1698 : // No data for request
1699 1 : buffer->push(2, 2);
1700 :
1701 1 : sts = fp.applyRequests(fps);
1702 4 : TRACE_REQUIRE_EQUAL(sts, true);
1703 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
1704 :
1705 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1706 :
1707 1 : sts = fp.applyRequests(fps);
1708 4 : TRACE_REQUIRE_EQUAL(sts, true);
1709 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1710 4 : while (fps.size() > 0)
1711 : {
1712 3 : ids[fps.front()->fragmentID()]++;
1713 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 2);
1714 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1715 3 : type = artdaq::Fragment::ContainerFragmentType;
1716 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1717 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1718 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1719 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1720 3 : type = artdaq::Fragment::FirstUserFragmentType;
1721 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1722 3 : fps.pop_front();
1723 3 : }
1724 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1725 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1726 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1727 1 : ids.clear();
1728 1 : fps.clear();
1729 :
1730 : // Request Timeout
1731 1 : buffer->push(4, 3);
1732 :
1733 1 : sts = fp.applyRequests(fps);
1734 4 : TRACE_REQUIRE_EQUAL(sts, true);
1735 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
1736 :
1737 1 : usleep(1500000);
1738 1 : sts = fp.applyRequests(fps);
1739 4 : TRACE_REQUIRE_EQUAL(sts, true);
1740 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1741 :
1742 : // Also, missing request timeout
1743 1 : auto list = fp.GetSentWindowList(1);
1744 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
1745 :
1746 1 : usleep(1500000);
1747 :
1748 4 : while (fps.size() > 0)
1749 : {
1750 3 : ids[fps.front()->fragmentID()]++;
1751 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 3);
1752 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
1753 3 : type = artdaq::Fragment::ContainerFragmentType;
1754 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1755 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1756 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 0);
1757 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), true);
1758 3 : type = artdaq::Fragment::EmptyFragmentType;
1759 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1760 3 : fps.pop_front();
1761 3 : }
1762 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1763 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1764 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1765 1 : ids.clear();
1766 :
1767 1 : fps.clear();
1768 :
1769 : // Data-taking has passed request
1770 1 : fp.AddFragmentsToBuffer(gen.Generate(12));
1771 :
1772 1 : buffer->push(5, 4);
1773 :
1774 1 : list = fp.GetSentWindowList(1); // Out-of-order list is only updated in getNext calls
1775 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
1776 1 : sts = fp.applyRequests(fps);
1777 1 : list = fp.GetSentWindowList(1);
1778 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
1779 4 : TRACE_REQUIRE_EQUAL(sts, true);
1780 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1781 4 : while (fps.size() > 0)
1782 : {
1783 3 : ids[fps.front()->fragmentID()]++;
1784 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 4);
1785 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 5);
1786 3 : type = artdaq::Fragment::ContainerFragmentType;
1787 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1788 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1789 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 0);
1790 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), true);
1791 3 : type = artdaq::Fragment::EmptyFragmentType;
1792 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1793 3 : fps.pop_front();
1794 3 : }
1795 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1796 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1797 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1798 1 : ids.clear();
1799 1 : fps.clear();
1800 :
1801 : // Out-of-order windows
1802 1 : buffer->push(7, 13);
1803 :
1804 1 : sts = fp.applyRequests(fps);
1805 4 : TRACE_REQUIRE_EQUAL(sts, true);
1806 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1807 4 : while (fps.size() > 0)
1808 : {
1809 3 : ids[fps.front()->fragmentID()]++;
1810 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 13);
1811 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 7);
1812 3 : type = artdaq::Fragment::ContainerFragmentType;
1813 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1814 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1815 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1816 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1817 3 : type = artdaq::Fragment::FirstUserFragmentType;
1818 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1819 3 : fps.pop_front();
1820 3 : }
1821 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1822 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1823 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1824 1 : ids.clear();
1825 1 : fps.clear();
1826 :
1827 1 : list = fp.GetSentWindowList(1);
1828 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
1829 :
1830 1 : buffer->push(6, 12);
1831 :
1832 1 : sts = fp.applyRequests(fps);
1833 4 : TRACE_REQUIRE_EQUAL(sts, true);
1834 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3);
1835 4 : while (fps.size() > 0)
1836 : {
1837 3 : ids[fps.front()->fragmentID()]++;
1838 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 12);
1839 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 6);
1840 3 : type = artdaq::Fragment::ContainerFragmentType;
1841 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1842 3 : auto cf = artdaq::ContainerFragment(*fps.front());
1843 12 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1844 12 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1845 3 : type = artdaq::Fragment::FirstUserFragmentType;
1846 12 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1847 3 : fps.pop_front();
1848 3 : }
1849 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1850 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1851 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1852 1 : ids.clear();
1853 1 : fps.clear();
1854 :
1855 1 : list = fp.GetSentWindowList(1);
1856 4 : TRACE_REQUIRE_EQUAL(list.size(), 0);
1857 :
1858 2 : fp.AddFragmentsToBuffer(gen.Generate(1, {1, 2}));
1859 :
1860 1 : buffer->push(8, 15);
1861 :
1862 1 : sts = fp.applyRequests(fps);
1863 4 : TRACE_REQUIRE_EQUAL(sts, true);
1864 4 : TRACE_REQUIRE_EQUAL(fps.size(), 2);
1865 3 : while (fps.size() > 0)
1866 : {
1867 2 : ids[fps.front()->fragmentID()]++;
1868 8 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 15);
1869 8 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 8);
1870 2 : type = artdaq::Fragment::ContainerFragmentType;
1871 8 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1872 2 : auto cf = artdaq::ContainerFragment(*fps.front());
1873 8 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1874 8 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1875 2 : type = artdaq::Fragment::FirstUserFragmentType;
1876 8 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1877 2 : fps.pop_front();
1878 2 : }
1879 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1880 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1881 4 : TRACE_REQUIRE_EQUAL(ids[3], 0);
1882 1 : ids.clear();
1883 1 : fps.clear();
1884 :
1885 1 : gen.setTimestamp(14); // Reset timestamp
1886 2 : fp.AddFragmentsToBuffer(gen.Generate(1, {3}));
1887 :
1888 1 : sts = fp.applyRequests(fps);
1889 4 : TRACE_REQUIRE_EQUAL(sts, true);
1890 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1);
1891 2 : while (fps.size() > 0)
1892 : {
1893 1 : ids[fps.front()->fragmentID()]++;
1894 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 15);
1895 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 8);
1896 1 : type = artdaq::Fragment::ContainerFragmentType;
1897 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1898 1 : auto cf = artdaq::ContainerFragment(*fps.front());
1899 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
1900 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
1901 1 : type = artdaq::Fragment::FirstUserFragmentType;
1902 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
1903 1 : fps.pop_front();
1904 1 : }
1905 4 : TRACE_REQUIRE_EQUAL(ids[1], 0);
1906 4 : TRACE_REQUIRE_EQUAL(ids[2], 0);
1907 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1908 1 : ids.clear();
1909 1 : fps.clear();
1910 :
1911 3 : TLOG(TLVL_INFO) << "WindowMode_Function_MultipleIDs test case END";
1912 1 : }
1913 :
1914 2 : BOOST_AUTO_TEST_CASE(SequenceIDMode_MultipleIDs)
1915 : {
1916 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
1917 3 : TLOG(TLVL_INFO) << "SequenceIDMode_MultipleIDs test case BEGIN";
1918 1 : fhicl::ParameterSet ps;
1919 4 : ps.put<std::vector<int>>("fragment_ids", {1, 2, 3});
1920 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
1921 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
1922 2 : ps.put<bool>("separate_data_thread", true);
1923 2 : ps.put<bool>("separate_monitoring_thread", false);
1924 2 : ps.put<bool>("receive_requests", true);
1925 3 : ps.put<int64_t>("hardware_poll_interval_us", 0);
1926 3 : ps.put<std::string>("request_mode", "SequenceID");
1927 :
1928 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
1929 1 : buffer->setRunning(true);
1930 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
1931 1 : artdaq::FragmentBuffer fp(ps);
1932 1 : fp.SetRequestBuffer(buffer);
1933 :
1934 1 : buffer->push(1, 1);
1935 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1936 :
1937 : // Test that Fragment with matching Sequence ID and timestamp is returned
1938 1 : artdaq::FragmentPtrs fps;
1939 1 : std::map<artdaq::Fragment::fragment_id_t, size_t> ids;
1940 1 : auto sts = fp.applyRequests(fps);
1941 1 : auto type = artdaq::Fragment::FirstUserFragmentType;
1942 4 : TRACE_REQUIRE_EQUAL(sts, true);
1943 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1944 :
1945 4 : for (auto ii = 0; ii < 3; ++ii)
1946 : {
1947 3 : ids[fps.front()->fragmentID()]++;
1948 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
1949 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
1950 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1951 3 : fps.pop_front();
1952 : }
1953 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1954 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1955 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1956 1 : ids.clear();
1957 :
1958 1 : fps.clear();
1959 :
1960 : // Test that no Fragment is returned when one does not exist in the buffer
1961 1 : buffer->push(2, 5);
1962 :
1963 1 : sts = fp.applyRequests(fps);
1964 4 : TRACE_REQUIRE_EQUAL(sts, true);
1965 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0u);
1966 :
1967 : // Test that Fragment with matching Sequence ID and non-matching timestamp is returned
1968 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
1969 :
1970 1 : sts = fp.applyRequests(fps);
1971 4 : TRACE_REQUIRE_EQUAL(sts, true);
1972 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1973 4 : for (auto ii = 0; ii < 3; ++ii)
1974 : {
1975 3 : ids[fps.front()->fragmentID()]++;
1976 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 2);
1977 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
1978 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
1979 3 : fps.pop_front();
1980 : }
1981 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
1982 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
1983 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
1984 1 : ids.clear();
1985 :
1986 1 : fps.clear();
1987 :
1988 : // Test out-of-order requests, with non-matching timestamps
1989 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
1990 :
1991 1 : buffer->push(4, 7);
1992 :
1993 1 : sts = fp.applyRequests(fps);
1994 4 : TRACE_REQUIRE_EQUAL(sts, true);
1995 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
1996 4 : for (auto ii = 0; ii < 3; ++ii)
1997 : {
1998 3 : ids[fps.front()->fragmentID()]++;
1999 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 4);
2000 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 4);
2001 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2002 3 : fps.pop_front();
2003 : }
2004 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
2005 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
2006 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
2007 1 : ids.clear();
2008 1 : fps.clear();
2009 :
2010 1 : buffer->push(3, 6);
2011 :
2012 1 : sts = fp.applyRequests(fps);
2013 4 : TRACE_REQUIRE_EQUAL(sts, true);
2014 4 : TRACE_REQUIRE_EQUAL(fps.size(), 3u);
2015 4 : for (auto ii = 0; ii < 3; ++ii)
2016 : {
2017 3 : ids[fps.front()->fragmentID()]++;
2018 12 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 3);
2019 12 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 3);
2020 12 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2021 3 : fps.pop_front();
2022 : }
2023 4 : TRACE_REQUIRE_EQUAL(ids[1], 1);
2024 4 : TRACE_REQUIRE_EQUAL(ids[2], 1);
2025 4 : TRACE_REQUIRE_EQUAL(ids[3], 1);
2026 1 : ids.clear();
2027 :
2028 3 : TLOG(TLVL_INFO) << "SequenceIDMode_MultipleIDs test case END";
2029 1 : }
2030 :
2031 2 : BOOST_AUTO_TEST_CASE(IgnoreRequests_StateMachine)
2032 : {
2033 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
2034 3 : TLOG(TLVL_INFO) << "IgnoreRequests_StateMachine test case BEGIN";
2035 1 : fhicl::ParameterSet ps;
2036 2 : ps.put<int>("fragment_id", 1);
2037 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
2038 3 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
2039 3 : ps.put<std::string>("request_mode", "ignored");
2040 :
2041 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
2042 1 : buffer->setRunning(true);
2043 :
2044 1 : artdaq::FragmentBuffer fp(ps);
2045 1 : fp.SetRequestBuffer(buffer);
2046 :
2047 1 : buffer->push(53, 35);
2048 :
2049 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
2050 :
2051 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
2052 :
2053 1 : artdaq::FragmentPtrs fps;
2054 1 : auto sts = fp.applyRequests(fps);
2055 :
2056 4 : TRACE_REQUIRE_EQUAL(sts, true);
2057 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
2058 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2059 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
2060 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
2061 1 : fps.clear();
2062 :
2063 1 : fp.Reset(false);
2064 1 : sts = fp.applyRequests(fps);
2065 :
2066 4 : TRACE_REQUIRE_EQUAL(sts, true);
2067 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0u);
2068 :
2069 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
2070 :
2071 1 : sts = fp.applyRequests(fps);
2072 :
2073 4 : TRACE_REQUIRE_EQUAL(sts, true);
2074 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
2075 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2076 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 2);
2077 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
2078 1 : fps.clear();
2079 :
2080 1 : fp.Stop();
2081 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
2082 :
2083 1 : sts = fp.applyRequests(fps);
2084 :
2085 4 : TRACE_REQUIRE_EQUAL(sts, false);
2086 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0u);
2087 :
2088 3 : TLOG(TLVL_INFO) << "IgnoreRequests_StateMachine test case END";
2089 1 : }
2090 :
2091 2 : BOOST_AUTO_TEST_CASE(SingleMode_StateMachine)
2092 : {
2093 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
2094 3 : TLOG(TLVL_INFO) << "SingleMode_StateMachine test case BEGIN";
2095 1 : fhicl::ParameterSet ps;
2096 2 : ps.put<int>("fragment_id", 1);
2097 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
2098 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
2099 3 : ps.put<bool>("receive_requests", true);
2100 3 : ps.put<std::string>("request_mode", "single");
2101 :
2102 1 : auto type = artdaq::Fragment::FirstUserFragmentType;
2103 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
2104 1 : buffer->setRunning(true);
2105 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
2106 1 : artdaq::FragmentBuffer fp(ps);
2107 1 : fp.SetRequestBuffer(buffer);
2108 :
2109 1 : buffer->push(1, 1);
2110 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
2111 :
2112 1 : artdaq::FragmentPtrs fps;
2113 1 : auto sts = fp.applyRequests(fps);
2114 4 : TRACE_REQUIRE_EQUAL(sts, true);
2115 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
2116 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2117 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
2118 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
2119 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2120 1 : fps.clear();
2121 :
2122 1 : fp.Reset(false);
2123 1 : buffer->reset();
2124 1 : sts = fp.applyRequests(fps);
2125 4 : TRACE_REQUIRE_EQUAL(sts, true);
2126 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0u);
2127 :
2128 1 : buffer->push(1, 1);
2129 1 : fp.AddFragmentsToBuffer(gen.Generate(1));
2130 1 : sts = fp.applyRequests(fps);
2131 4 : TRACE_REQUIRE_EQUAL(sts, true);
2132 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
2133 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2134 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
2135 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
2136 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2137 1 : fps.clear();
2138 :
2139 1 : fp.Stop();
2140 1 : buffer->push(2, 5);
2141 1 : sts = fp.applyRequests(fps);
2142 4 : TRACE_REQUIRE_EQUAL(sts, true);
2143 4 : TRACE_REQUIRE_EQUAL(fps.size(), 1u);
2144 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2145 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 5);
2146 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 2);
2147 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2148 1 : fps.clear();
2149 :
2150 1 : buffer->setRunning(false);
2151 1 : fp.AddFragmentsToBuffer(gen.Generate(2));
2152 1 : buffer->push(4, 7);
2153 :
2154 1 : sts = fp.applyRequests(fps);
2155 4 : TRACE_REQUIRE_EQUAL(sts, false);
2156 4 : TRACE_REQUIRE_EQUAL(fps.size(), 0);
2157 :
2158 3 : TLOG(TLVL_INFO) << "SingleMode_StateMachine test case END";
2159 1 : }
2160 :
2161 2 : BOOST_AUTO_TEST_CASE(WindowMode_RateTests)
2162 : {
2163 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
2164 3 : TLOG(TLVL_INFO) << "WindowMode_RateTests test case BEGIN";
2165 1 : fhicl::ParameterSet ps;
2166 2 : ps.put<int>("fragment_id", 1);
2167 :
2168 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
2169 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
2170 2 : ps.put<size_t>("data_buffer_depth_fragments", 2 * RATE_TEST_COUNT);
2171 2 : ps.put<bool>("circular_buffer_mode", false);
2172 3 : ps.put<bool>("receive_requests", true);
2173 3 : ps.put<std::string>("request_mode", "window");
2174 2 : ps.put<size_t>("missing_request_window_timeout_us", 500000);
2175 2 : ps.put<size_t>("window_close_timeout_us", 500000);
2176 :
2177 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
2178 1 : buffer->setRunning(true);
2179 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
2180 1 : artdaq::FragmentBuffer fp(ps);
2181 1 : fp.SetRequestBuffer(buffer);
2182 :
2183 1 : auto beginop = std::chrono::steady_clock::now();
2184 :
2185 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests BEGIN";
2186 1 : size_t req_seq = 1;
2187 100001 : for (; req_seq < RATE_TEST_COUNT + 1; ++req_seq)
2188 : {
2189 100000 : buffer->push(req_seq, req_seq);
2190 : }
2191 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests END.Time elapsed = " << artdaq::TimeUtils::GetElapsedTime(beginop)
2192 2 : << "(" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " reqs/s) ";
2193 :
2194 1 : beginop = std::chrono::steady_clock::now();
2195 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments BEGIN";
2196 1 : fp.AddFragmentsToBuffer(gen.Generate(RATE_TEST_COUNT));
2197 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop)
2198 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " frags/s)";
2199 :
2200 1 : beginop = std::chrono::steady_clock::now();
2201 3 : TLOG(TLVL_INFO) << "Applying requests BEGIN";
2202 1 : artdaq::FragmentPtrs fps;
2203 1 : auto sts = fp.applyRequests(fps);
2204 3 : TLOG(TLVL_INFO) << "Applying requests END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop)
2205 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " reqs/s)";
2206 4 : TRACE_REQUIRE_EQUAL(sts, true);
2207 4 : TRACE_REQUIRE_EQUAL(fps.size(), RATE_TEST_COUNT);
2208 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2209 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
2210 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
2211 1 : auto type = artdaq::Fragment::ContainerFragmentType;
2212 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2213 1 : BOOST_REQUIRE_GE(fps.front()->sizeBytes(), 2 * sizeof(artdaq::detail::RawFragmentHeader) + sizeof(artdaq::ContainerFragment::Metadata));
2214 1 : auto cf = artdaq::ContainerFragment(*fps.front());
2215 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 1);
2216 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), false);
2217 1 : type = artdaq::Fragment::FirstUserFragmentType;
2218 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
2219 1 : fps.clear();
2220 :
2221 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests BEGIN";
2222 100001 : for (; req_seq < (2 * RATE_TEST_COUNT) + 1; ++req_seq)
2223 : {
2224 100000 : buffer->push(req_seq, req_seq);
2225 : }
2226 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests END. Time elapsed = " << artdaq::TimeUtils::GetElapsedTime(beginop)
2227 2 : << "(" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " reqs/s) ";
2228 :
2229 1 : beginop = std::chrono::steady_clock::now();
2230 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments Individually BEGIN";
2231 100001 : for (int ii = 0; ii < RATE_TEST_COUNT; ++ii)
2232 100000 : fp.AddFragmentsToBuffer(gen.Generate(1));
2233 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments Individually END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop)
2234 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " frags/s)";
2235 :
2236 3 : TLOG(TLVL_INFO) << "WindowMode_RateTests test case END";
2237 1 : }
2238 :
2239 2 : BOOST_AUTO_TEST_CASE(CircularBufferMode_RateTests)
2240 : {
2241 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
2242 3 : TLOG(TLVL_INFO) << "CircularBufferMode_RateTests test case BEGIN";
2243 1 : fhicl::ParameterSet ps;
2244 2 : ps.put<int>("fragment_id", 1);
2245 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
2246 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
2247 2 : ps.put<size_t>("data_buffer_depth_fragments", RATE_TEST_COUNT / 2);
2248 2 : ps.put<bool>("circular_buffer_mode", true);
2249 3 : ps.put<bool>("receive_requests", true);
2250 3 : ps.put<std::string>("request_mode", "window");
2251 2 : ps.put<size_t>("missing_request_window_timeout_us", 500000);
2252 2 : ps.put<size_t>("window_close_timeout_us", 500000);
2253 :
2254 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
2255 1 : buffer->setRunning(true);
2256 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
2257 1 : artdaq::FragmentBuffer fp(ps);
2258 1 : fp.SetRequestBuffer(buffer);
2259 :
2260 1 : auto beginop = std::chrono::steady_clock::now();
2261 :
2262 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests BEGIN";
2263 100001 : for (size_t ii = 1; ii < RATE_TEST_COUNT + 1; ++ii)
2264 : {
2265 100000 : buffer->push(ii, ii);
2266 : }
2267 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests END. Time elapsed = " << artdaq::TimeUtils::GetElapsedTime(beginop)
2268 2 : << "(" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " reqs / s) ";
2269 :
2270 1 : beginop = std::chrono::steady_clock::now();
2271 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments BEGIN";
2272 1 : fp.AddFragmentsToBuffer(gen.Generate(RATE_TEST_COUNT));
2273 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop)
2274 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " frags/s)";
2275 :
2276 1 : beginop = std::chrono::steady_clock::now();
2277 3 : TLOG(TLVL_INFO) << "Applying requests BEGIN";
2278 1 : artdaq::FragmentPtrs fps;
2279 1 : auto sts = fp.applyRequests(fps);
2280 3 : TLOG(TLVL_INFO) << "Applying requests END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop)
2281 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " reqs/s)";
2282 4 : TRACE_REQUIRE_EQUAL(sts, true);
2283 4 : TRACE_REQUIRE_EQUAL(fps.size(), RATE_TEST_COUNT);
2284 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2285 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
2286 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
2287 1 : auto type = artdaq::Fragment::ContainerFragmentType;
2288 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2289 1 : auto cf = artdaq::ContainerFragment(*fps.front());
2290 4 : TRACE_REQUIRE_EQUAL(cf.block_count(), 0);
2291 4 : TRACE_REQUIRE_EQUAL(cf.missing_data(), true);
2292 1 : type = artdaq::Fragment::EmptyFragmentType;
2293 4 : TRACE_REQUIRE_EQUAL(cf.fragment_type(), type);
2294 1 : fps.clear();
2295 :
2296 3 : TLOG(TLVL_INFO) << "CircularBufferMode_RateTests test case END";
2297 1 : }
2298 :
2299 2 : BOOST_AUTO_TEST_CASE(WindowMode_RateTests_threaded)
2300 : {
2301 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
2302 3 : TLOG(TLVL_INFO) << "WindowMode_RateTests_threaded test case BEGIN";
2303 1 : fhicl::ParameterSet ps;
2304 2 : ps.put<int>("fragment_id", 1);
2305 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
2306 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
2307 2 : ps.put<size_t>("data_buffer_depth_fragments", RATE_TEST_COUNT);
2308 3 : ps.put<bool>("circular_buffer_mode", false);
2309 3 : ps.put<std::string>("request_mode", "window");
2310 2 : ps.put<bool>("receive_requests", true);
2311 2 : ps.put<size_t>("missing_request_window_timeout_us", 500000);
2312 2 : ps.put<size_t>("window_close_timeout_us", 500000);
2313 :
2314 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
2315 1 : buffer->setRunning(true);
2316 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
2317 1 : artdaq::FragmentBuffer fp(ps);
2318 1 : fp.SetRequestBuffer(buffer);
2319 :
2320 1 : std::atomic<int> thread_sync = 0;
2321 1 : auto gen_requests = [&]() {
2322 1 : ++thread_sync;
2323 25631 : while (thread_sync < 3) {}
2324 1 : auto beginop = std::chrono::steady_clock::now();
2325 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests BEGIN";
2326 100001 : for (size_t ii = 1; ii < RATE_TEST_COUNT + 1; ++ii)
2327 : {
2328 100000 : buffer->push(ii, ii);
2329 : }
2330 3 : TLOG(TLVL_INFO) << "Generating " << RATE_TEST_COUNT << " requests END. Time elapsed = " << artdaq::TimeUtils::GetElapsedTime(beginop)
2331 2 : << "(" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " reqs / s) ";
2332 2 : };
2333 :
2334 1 : auto gen_frags = [&]() {
2335 1 : ++thread_sync;
2336 895 : while (thread_sync < 3) {}
2337 1 : auto beginop = std::chrono::steady_clock::now();
2338 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments BEGIN";
2339 100001 : for (int ii = 0; ii < RATE_TEST_COUNT; ++ii)
2340 100000 : fp.AddFragmentsToBuffer(gen.Generate(1));
2341 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop)
2342 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " frags/s)";
2343 2 : };
2344 :
2345 1 : auto apply_requests = [&]() {
2346 1 : ++thread_sync;
2347 1 : while (thread_sync < 3) {}
2348 1 : artdaq::FragmentPtrs fps;
2349 1 : auto begin_test = std::chrono::steady_clock::now();
2350 3 : while (fps.size() < RATE_TEST_COUNT && artdaq::TimeUtils::GetElapsedTime(begin_test) < RATE_TEST_COUNT / 1000) // 1 ms per request
2351 : {
2352 2 : auto beginop = std::chrono::steady_clock::now();
2353 6 : TLOG(TLVL_INFO) << "Applying requests BEGIN";
2354 2 : auto sts = fp.applyRequests(fps);
2355 8 : TRACE_REQUIRE_EQUAL(sts, true);
2356 6 : TLOG(TLVL_INFO) << "Applying requests END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop);
2357 : }
2358 1 : if (fps.size() < RATE_TEST_COUNT)
2359 : {
2360 0 : TLOG(TLVL_WARNING) << "Some requests did not return data. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(begin_test) << ", " << fps.size() << " / " << RATE_TEST_COUNT << " Fragments received";
2361 : }
2362 : else
2363 : {
2364 3 : TLOG(TLVL_INFO) << "All request replies received. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(begin_test)
2365 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(begin_test) << " reqs/s)";
2366 :
2367 4 : TRACE_REQUIRE_EQUAL(fps.size(), RATE_TEST_COUNT);
2368 : }
2369 4 : TRACE_REQUIRE_EQUAL(fps.front()->fragmentID(), 1);
2370 4 : TRACE_REQUIRE_EQUAL(fps.front()->timestamp(), 1);
2371 4 : TRACE_REQUIRE_EQUAL(fps.front()->sequenceID(), 1);
2372 1 : auto type = artdaq::Fragment::ContainerFragmentType;
2373 4 : TRACE_REQUIRE_EQUAL(fps.front()->type(), type);
2374 1 : fps.clear();
2375 2 : };
2376 :
2377 1 : std::thread t1(gen_requests);
2378 1 : std::thread t2(gen_frags);
2379 1 : std::thread t3(apply_requests);
2380 :
2381 1 : t1.join();
2382 1 : t2.join();
2383 1 : t3.join();
2384 :
2385 3 : TLOG(TLVL_INFO) << "WindowMode_RateTests_threaded test case END";
2386 1 : }
2387 :
2388 2 : BOOST_AUTO_TEST_CASE(WaitForDataBufferReady_RaceCondition)
2389 : {
2390 1 : artdaq::configureMessageFacility("FragmentBuffer_t", true, MESSAGEFACILITY_DEBUG);
2391 3 : TLOG(TLVL_INFO) << "WaitForDataBufferReady_RaceCondition test case BEGIN";
2392 1 : fhicl::ParameterSet ps;
2393 2 : ps.put<int>("fragment_id", 1);
2394 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_offset", 0);
2395 2 : ps.put<artdaq::Fragment::timestamp_t>("request_window_width", 0);
2396 2 : ps.put<size_t>("data_buffer_depth_fragments", 1);
2397 2 : ps.put<bool>("circular_buffer_mode", false);
2398 3 : ps.put<bool>("receive_requests", true);
2399 3 : ps.put<std::string>("request_mode", "window");
2400 :
2401 1 : auto buffer = std::make_shared<artdaq::RequestBuffer>();
2402 1 : buffer->setRunning(true);
2403 1 : artdaqtest::FragmentBufferTestGenerator gen(ps);
2404 1 : artdaq::FragmentBuffer fp(ps);
2405 1 : fp.SetRequestBuffer(buffer);
2406 :
2407 1 : std::atomic<int> thread_sync = 0;
2408 :
2409 1 : auto gen_frags = [&]() {
2410 1 : ++thread_sync;
2411 12495 : while (thread_sync < 2) {}
2412 1 : auto beginop = std::chrono::steady_clock::now();
2413 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments BEGIN";
2414 100001 : for (int ii = 0; ii < RATE_TEST_COUNT; ++ii)
2415 100000 : fp.AddFragmentsToBuffer(gen.Generate(1));
2416 3 : TLOG(TLVL_INFO) << "Generating/adding " << RATE_TEST_COUNT << " Fragments END. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop)
2417 2 : << " (" << RATE_TEST_COUNT / artdaq::TimeUtils::GetElapsedTime(beginop) << " frags/s)";
2418 1 : thread_sync = 0;
2419 2 : };
2420 :
2421 1 : auto reset_buffer = [&]() {
2422 1 : ++thread_sync;
2423 1 : while (thread_sync < 2) {}
2424 1 : auto beginop = std::chrono::steady_clock::now();
2425 3 : TLOG(TLVL_INFO) << "Resetting loop BEGIN";
2426 1 : size_t counter = 0;
2427 18946243 : while (thread_sync > 0)
2428 : {
2429 18946242 : fp.Reset(false);
2430 18946242 : counter++;
2431 : }
2432 3 : TLOG(TLVL_INFO) << "Reset " << counter << " times during loop. Time elapsed=" << artdaq::TimeUtils::GetElapsedTime(beginop);
2433 2 : };
2434 :
2435 1 : std::thread t1(gen_frags);
2436 1 : std::thread t2(reset_buffer);
2437 :
2438 1 : t1.join();
2439 1 : t2.join();
2440 :
2441 3 : TLOG(TLVL_INFO) << "WaitForDataBufferReady_RaceCondition test case END";
2442 1 : }
2443 :
2444 : BOOST_AUTO_TEST_SUITE_END()
|
