Line data Source code
1 : /**
2 : * @file SenderSafety_test.cxx
3 : *
4 : * Failure-injection tests for the sender-safety components: TX engine
5 : * ownership, the callback-lifetime gate, and the WIBEthTransmitter core
6 : * (per-stream accounting, registry callbacks, observer hook). All tests run
7 : * without EAL or a NIC.
8 : */
9 :
10 : #include "dpdklibs/sender/CallbackGate.hpp"
11 : #include "dpdklibs/sender/TxEngine.hpp"
12 : #include "dpdklibs/sender/WIBEthTransmitter.hpp"
13 :
14 : #define BOOST_TEST_MODULE SenderSafety_test // NOLINT
15 :
16 : #include "boost/test/unit_test.hpp"
17 :
18 : #include <array>
19 : #include <atomic>
20 : #include <chrono>
21 : #include <cstdint>
22 : #include <cstring>
23 : #include <functional>
24 : #include <future>
25 : #include <map>
26 : #include <string>
27 : #include <thread>
28 : #include <vector>
29 :
30 : using namespace dunedaq::dpdklibs;
31 :
32 : namespace {
33 :
34 : // Deterministic scriptable backend implementing the TxEngine concept.
35 : struct FakeBackend
36 : {
37 : int alloc_calls = 0;
38 : int prepare_calls = 0;
39 : int tx_calls = 0;
40 : int free_calls = 0;
41 :
42 : bool fail_alloc = false;
43 : bool fail_prepare = false;
44 : bool fail_tx = false;
45 :
46 : std::array<std::uint8_t, 64> storage{};
47 :
48 5 : void* alloc()
49 : {
50 5 : ++alloc_calls;
51 5 : return fail_alloc ? nullptr : storage.data();
52 : }
53 :
54 4 : std::uint8_t* prepare(void* buf, std::size_t /*bytes*/)
55 : {
56 4 : ++prepare_calls;
57 4 : return fail_prepare ? nullptr : static_cast<std::uint8_t*>(buf);
58 : }
59 :
60 3 : bool transmit(void* /*buf*/)
61 : {
62 3 : ++tx_calls;
63 3 : return !fail_tx;
64 : }
65 :
66 3 : void free(void* /*buf*/) { ++free_calls; }
67 : };
68 :
69 : constexpr std::size_t kTestPacketBytes = 16;
70 :
71 : // ---------------------------------------------------------------------------
72 : // WIBEthTransmitter fixtures
73 : // ---------------------------------------------------------------------------
74 :
75 : // Payload satisfying the transmitter's contract (a `data` member addressing
76 : // one complete WIBEthFrame), mirroring DUNEWIBEthTypeAdapter's layout.
77 : struct FakePayload
78 : {
79 : char data[wibeth::kWIBEthFrameBytes] = {};
80 : };
81 :
82 : // Backend with a full-packet capture buffer, so tests can inspect the bytes
83 : // the transmitter passed to transmit().
84 : struct FakeTxBackend
85 : {
86 : int alloc_calls = 0;
87 : int tx_calls = 0;
88 : int free_calls = 0;
89 : bool fail_transmit = false;
90 : std::array<std::uint8_t, wibeth::kEthernetPacketBytes> storage{};
91 :
92 7 : void* alloc()
93 : {
94 7 : ++alloc_calls;
95 7 : return storage.data();
96 : }
97 7 : std::uint8_t* prepare(void* buf, std::size_t /*bytes*/) { return static_cast<std::uint8_t*>(buf); }
98 7 : bool transmit(void* /*buf*/)
99 : {
100 7 : ++tx_calls;
101 7 : return !fail_transmit;
102 : }
103 1 : void free(void* /*buf*/) { ++free_calls; }
104 : };
105 :
106 : struct FakeCallbackConf
107 : {
108 : std::string uid;
109 10 : const std::string& UID() const { return uid; }
110 : };
111 :
112 : // Stand-in for the pinned DataMoveCallbackRegistry: identical
113 : // register_callback signature, keyed by conf->UID(). The first registration
114 : // for an ID is retained.
115 : struct FakeCallbackRegistry
116 : {
117 : std::map<std::string, std::function<void(FakePayload&&)>> callbacks;
118 : int rejected_duplicates = 0;
119 :
120 : template<typename DataType>
121 5 : void register_callback(const FakeCallbackConf* conf, std::function<void(DataType&&)> callback)
122 : {
123 5 : if (callbacks.count(conf->UID()) != 0) {
124 0 : ++rejected_duplicates;
125 0 : return;
126 : }
127 5 : callbacks[conf->UID()] = std::move(callback);
128 : }
129 : };
130 :
131 : using test_transmitter_t = sender::WIBEthTransmitter<FakeTxBackend>;
132 :
133 : dunedaq::detdataformats::DAQEthHeader
134 3 : read_wire_daq_header(const std::array<std::uint8_t, wibeth::kEthernetPacketBytes>& packet)
135 : {
136 3 : dunedaq::detdataformats::DAQEthHeader header;
137 3 : std::memcpy(&header, packet.data() + wibeth::kPacketHeaderBytes, sizeof(header));
138 3 : return header;
139 : }
140 :
141 : } // namespace
142 :
143 : BOOST_AUTO_TEST_SUITE(SenderSafety_test)
144 :
145 : // ---------------------------------------------------------------------------
146 : // TxEngine ownership and failure matrix (F03/F08)
147 : // ---------------------------------------------------------------------------
148 :
149 2 : BOOST_AUTO_TEST_CASE(EngineAllocFailureFreesNothingAndSkipsBuild)
150 : {
151 1 : FakeBackend backend;
152 1 : backend.fail_alloc = true;
153 1 : bool built = false;
154 1 : bool sent = false;
155 :
156 2 : const auto outcome = sender::send_packet(
157 1 : backend, kTestPacketBytes, nullptr, [&](std::uint8_t*) { built = true; }, [&]() { sent = true; });
158 :
159 1 : BOOST_CHECK(outcome == sender::TxOutcome::kAllocFailed);
160 1 : BOOST_CHECK(!built);
161 1 : BOOST_CHECK(!sent);
162 1 : BOOST_CHECK_EQUAL(backend.prepare_calls, 0);
163 1 : BOOST_CHECK_EQUAL(backend.tx_calls, 0);
164 1 : BOOST_CHECK_EQUAL(backend.free_calls, 0);
165 1 : }
166 :
167 2 : BOOST_AUTO_TEST_CASE(EnginePrepareFailureFreesExactlyOnce)
168 : {
169 1 : FakeBackend backend;
170 1 : backend.fail_prepare = true;
171 1 : bool built = false;
172 :
173 1 : const auto outcome =
174 1 : sender::send_packet(backend, kTestPacketBytes, nullptr, [&](std::uint8_t*) { built = true; }, []() {});
175 :
176 1 : BOOST_CHECK(outcome == sender::TxOutcome::kPrepareFailed);
177 1 : BOOST_CHECK(!built);
178 1 : BOOST_CHECK_EQUAL(backend.tx_calls, 0);
179 1 : BOOST_CHECK_EQUAL(backend.free_calls, 1);
180 1 : }
181 :
182 2 : BOOST_AUTO_TEST_CASE(EngineSuccessTransfersOwnership)
183 : {
184 1 : FakeBackend backend;
185 1 : int sent_count = 0;
186 :
187 2 : const auto outcome = sender::send_packet(
188 3 : backend, kTestPacketBytes, nullptr, [](std::uint8_t* dst) { dst[0] = 0xab; }, [&]() { ++sent_count; });
189 :
190 1 : BOOST_CHECK(outcome == sender::TxOutcome::kSent);
191 1 : BOOST_CHECK_EQUAL(sent_count, 1);
192 1 : BOOST_CHECK_EQUAL(backend.tx_calls, 1);
193 1 : BOOST_CHECK_EQUAL(backend.free_calls, 0); // NIC owns the buffer
194 1 : BOOST_CHECK_EQUAL(backend.storage[0], 0xab);
195 1 : }
196 :
197 2 : BOOST_AUTO_TEST_CASE(EngineBackpressureIsOneAttemptAndFreesExactlyOnce)
198 : {
199 1 : FakeBackend backend;
200 1 : backend.fail_tx = true;
201 1 : bool sent = false;
202 :
203 1 : const auto outcome =
204 1 : sender::send_packet(backend, kTestPacketBytes, nullptr, [](std::uint8_t*) {}, [&]() { sent = true; });
205 :
206 1 : BOOST_CHECK(outcome == sender::TxOutcome::kTxFailed);
207 1 : BOOST_CHECK(!sent);
208 1 : BOOST_CHECK_EQUAL(backend.tx_calls, 1); // refused once, recorded, not retried
209 1 : BOOST_CHECK_EQUAL(backend.free_calls, 1);
210 1 : }
211 :
212 2 : BOOST_AUTO_TEST_CASE(EnginePreTxCopyCapturesConstructedBytes)
213 : {
214 1 : FakeBackend backend;
215 1 : backend.fail_tx = true; // even on failure the copy must exist
216 1 : std::array<std::uint8_t, kTestPacketBytes> copy{};
217 :
218 1 : const auto outcome = sender::send_packet(
219 : backend,
220 : kTestPacketBytes,
221 : copy.data(),
222 2 : [](std::uint8_t* dst) {
223 17 : for (std::size_t i = 0; i < kTestPacketBytes; ++i) {
224 16 : dst[i] = static_cast<std::uint8_t>(i + 1);
225 : }
226 1 : },
227 1 : []() {});
228 :
229 1 : BOOST_CHECK(outcome == sender::TxOutcome::kTxFailed);
230 17 : for (std::size_t i = 0; i < kTestPacketBytes; ++i) {
231 16 : BOOST_CHECK_EQUAL(copy[i], static_cast<std::uint8_t>(i + 1));
232 : }
233 1 : }
234 :
235 : // ---------------------------------------------------------------------------
236 : // CallbackGate lifetime contract (F09)
237 : // ---------------------------------------------------------------------------
238 :
239 2 : BOOST_AUTO_TEST_CASE(GateDispatchesOnlyWhenArmed)
240 : {
241 1 : sender::CallbackGate<int> gate;
242 1 : int owner = 42;
243 :
244 1 : int calls = 0;
245 1 : BOOST_CHECK(!gate.dispatch([&](int&) { ++calls; })); // unarmed
246 1 : BOOST_CHECK_EQUAL(calls, 0);
247 :
248 1 : gate.arm(&owner);
249 1 : BOOST_CHECK(gate.armed());
250 2 : BOOST_CHECK(gate.dispatch([&](int& value) {
251 : ++calls;
252 : BOOST_CHECK_EQUAL(value, 42);
253 : }));
254 1 : BOOST_CHECK_EQUAL(calls, 1);
255 :
256 1 : gate.tombstone();
257 1 : BOOST_CHECK(!gate.armed());
258 1 : BOOST_CHECK(!gate.dispatch([&](int&) { ++calls; }));
259 1 : BOOST_CHECK_EQUAL(calls, 1);
260 :
261 : // Re-arm (configure after scrap on the same instance) restores dispatch.
262 1 : gate.arm(&owner);
263 2 : BOOST_CHECK(gate.dispatch([&](int&) { ++calls; }));
264 1 : BOOST_CHECK_EQUAL(calls, 2);
265 1 : }
266 :
267 2 : BOOST_AUTO_TEST_CASE(GateTombstoneDrainsInFlightDispatch)
268 : {
269 1 : sender::CallbackGate<int> gate;
270 1 : int owner = 7;
271 1 : gate.arm(&owner);
272 :
273 : // Bounded coordination: the dispatch signals entry, then blocks on a release
274 : // future. A second thread calls tombstone(), which must not return while the
275 : // dispatch is inside the gate.
276 1 : std::promise<void> entered_promise;
277 1 : auto entered = entered_promise.get_future();
278 1 : std::promise<void> release_promise;
279 1 : auto release = release_promise.get_future().share();
280 :
281 1 : std::atomic<bool> dispatch_finished{ false };
282 1 : std::atomic<bool> tombstone_returned{ false };
283 :
284 3 : std::thread worker([&]() {
285 1 : gate.dispatch([&](int&) {
286 1 : entered_promise.set_value();
287 1 : release.wait();
288 1 : dispatch_finished.store(true);
289 1 : });
290 2 : });
291 :
292 1 : BOOST_REQUIRE(entered.wait_for(std::chrono::seconds(5)) == std::future_status::ready);
293 :
294 3 : std::thread tombstoner([&]() {
295 1 : gate.tombstone();
296 1 : tombstone_returned.store(true);
297 2 : });
298 :
299 : // Allow the second thread time to reach the writer lock. It must remain
300 : // blocked while the dispatch is held open.
301 1 : std::this_thread::sleep_for(std::chrono::milliseconds(50));
302 1 : BOOST_CHECK(!tombstone_returned.load());
303 :
304 : // Release the dispatch. tombstone() must then complete.
305 1 : release_promise.set_value();
306 1 : worker.join();
307 1 : tombstoner.join();
308 :
309 1 : BOOST_CHECK(dispatch_finished.load());
310 1 : BOOST_CHECK(tombstone_returned.load());
311 1 : BOOST_CHECK(!gate.armed());
312 1 : }
313 :
314 : // ---------------------------------------------------------------------------
315 : // WIBEthTransmitter core: registry callbacks, accounting, observer hook
316 : // ---------------------------------------------------------------------------
317 :
318 2 : BOOST_AUTO_TEST_CASE(TransmitterRegistryPathBuildsAndSends)
319 : {
320 1 : test_transmitter_t tx;
321 1 : tx.add_stream("stream-a", 0);
322 1 : tx.add_stream("stream-b", 100);
323 1 : tx.configure(wibeth::PacketConfig{}, 3, FakeTxBackend{});
324 :
325 1 : FakeCallbackRegistry registry;
326 1 : FakeCallbackConf conf_a{ "stream-a" };
327 1 : FakeCallbackConf conf_b{ "stream-b" };
328 1 : const std::vector<const FakeCallbackConf*> confs{ &conf_a, &conf_b };
329 1 : tx.register_streams<FakePayload>(registry, confs);
330 1 : BOOST_REQUIRE_EQUAL(registry.callbacks.size(), 2);
331 1 : BOOST_CHECK(tx.armed());
332 :
333 1 : tx.set_accepting(true);
334 1 : FakePayload payload;
335 7201 : for (std::size_t i = 0; i < sizeof(payload.data); ++i) {
336 7200 : payload.data[i] = static_cast<char>(i % 251);
337 : }
338 1 : registry.callbacks.at("stream-a")(FakePayload(payload));
339 :
340 1 : const auto stats = tx.snapshot_stats();
341 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").accepted, 1);
342 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").sent, 1);
343 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").next_seq, 1);
344 1 : BOOST_CHECK_EQUAL(stats.at("stream-b").accepted, 0);
345 1 : BOOST_CHECK_EQUAL(tx.backend().tx_calls, 1);
346 1 : BOOST_CHECK_EQUAL(tx.backend().free_calls, 0); // NIC owns the buffer
347 :
348 : // Payload bytes appear unchanged after the patched DAQEthHeader. The patch
349 : // carries det_id and the stream sequence number.
350 1 : const auto& packet = tx.backend().storage;
351 1 : BOOST_CHECK(std::memcmp(packet.data() + wibeth::kPacketHeaderBytes + wibeth::kDAQEthHeaderBytes,
352 : payload.data + wibeth::kDAQEthHeaderBytes,
353 : wibeth::kWIBEthFrameBytes - wibeth::kDAQEthHeaderBytes) == 0);
354 1 : const auto header = read_wire_daq_header(packet);
355 1 : BOOST_CHECK_EQUAL(header.det_id, 3);
356 1 : BOOST_CHECK_EQUAL(header.seq_id, 0);
357 1 : }
358 :
359 2 : BOOST_AUTO_TEST_CASE(TransmitterUnknownStreamIsCountedDrop)
360 : {
361 1 : test_transmitter_t tx;
362 1 : tx.add_stream("known", 0);
363 1 : tx.configure(wibeth::PacketConfig{}, 0, FakeTxBackend{});
364 1 : tx.set_accepting(true);
365 :
366 1 : tx.transmit("mystery", FakePayload{});
367 :
368 1 : BOOST_CHECK_EQUAL(tx.unknown_stream_drops(), 1);
369 1 : BOOST_CHECK_EQUAL(tx.snapshot_stats().at("known").accepted, 0);
370 1 : BOOST_CHECK_EQUAL(tx.backend().tx_calls, 0);
371 1 : }
372 :
373 2 : BOOST_AUTO_TEST_CASE(TransmitterDropsWhileNotAccepting)
374 : {
375 1 : test_transmitter_t tx;
376 1 : tx.add_stream("stream-a", 0);
377 1 : tx.configure(wibeth::PacketConfig{}, 0, FakeTxBackend{});
378 :
379 1 : FakeCallbackRegistry registry;
380 1 : FakeCallbackConf conf_a{ "stream-a" };
381 1 : tx.register_streams<FakePayload>(registry, std::vector<const FakeCallbackConf*>{ &conf_a });
382 :
383 : // accepting defaults to false (pre-start / post-stop)
384 1 : registry.callbacks.at("stream-a")(FakePayload{});
385 :
386 1 : const auto stats = tx.snapshot_stats();
387 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").accepted, 1);
388 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").dropped_not_running, 1);
389 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").sent, 0);
390 1 : BOOST_CHECK_EQUAL(tx.backend().tx_calls, 0);
391 1 : }
392 :
393 2 : BOOST_AUTO_TEST_CASE(TransmitterSequenceWrapsModulo4096)
394 : {
395 1 : test_transmitter_t tx;
396 1 : tx.add_stream("stream-a", 4095);
397 1 : tx.configure(wibeth::PacketConfig{}, 0, FakeTxBackend{});
398 1 : tx.set_accepting(true);
399 :
400 1 : tx.transmit("stream-a", FakePayload{});
401 1 : BOOST_CHECK_EQUAL(read_wire_daq_header(tx.backend().storage).seq_id, 4095);
402 :
403 1 : tx.transmit("stream-a", FakePayload{});
404 1 : BOOST_CHECK_EQUAL(read_wire_daq_header(tx.backend().storage).seq_id, 0);
405 :
406 1 : const auto stats = tx.snapshot_stats();
407 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").sent, 2);
408 1 : BOOST_CHECK_EQUAL(stats.at("stream-a").next_seq, 1);
409 1 : }
410 :
411 2 : BOOST_AUTO_TEST_CASE(TransmitterTombstoneDrainsRegistryLambdaToSafeDrop)
412 : {
413 1 : test_transmitter_t tx;
414 1 : tx.add_stream("stream-a", 0);
415 1 : tx.configure(wibeth::PacketConfig{}, 0, FakeTxBackend{});
416 :
417 1 : FakeCallbackRegistry registry;
418 1 : FakeCallbackConf conf_a{ "stream-a" };
419 1 : tx.register_streams<FakePayload>(registry, std::vector<const FakeCallbackConf*>{ &conf_a });
420 1 : tx.set_accepting(true);
421 :
422 1 : registry.callbacks.at("stream-a")(FakePayload{});
423 1 : BOOST_CHECK_EQUAL(tx.snapshot_stats().at("stream-a").sent, 1);
424 :
425 1 : tx.tombstone();
426 1 : BOOST_CHECK(!tx.armed());
427 :
428 : // The registry still holds the lambda. Dispatch becomes a drop: no
429 : // accounting and no backend call.
430 1 : registry.callbacks.at("stream-a")(FakePayload{});
431 1 : BOOST_CHECK_EQUAL(tx.snapshot_stats().at("stream-a").accepted, 1);
432 1 : BOOST_CHECK_EQUAL(tx.backend().tx_calls, 1);
433 1 : }
434 :
435 2 : BOOST_AUTO_TEST_CASE(TransmitterReRegisterReArmsWithoutDuplicateRegistration)
436 : {
437 1 : test_transmitter_t tx;
438 1 : tx.add_stream("stream-a", 0);
439 1 : tx.configure(wibeth::PacketConfig{}, 0, FakeTxBackend{});
440 :
441 1 : FakeCallbackRegistry registry;
442 1 : FakeCallbackConf conf_a{ "stream-a" };
443 1 : const std::vector<const FakeCallbackConf*> confs{ &conf_a };
444 1 : tx.register_streams<FakePayload>(registry, confs);
445 1 : tx.tombstone();
446 :
447 : // configure after scrap on the same instance: re-arm only. The pinned
448 : // registry would reject a second registration, so none is attempted.
449 1 : tx.register_streams<FakePayload>(registry, confs);
450 1 : BOOST_CHECK_EQUAL(registry.callbacks.size(), 1);
451 1 : BOOST_CHECK_EQUAL(registry.rejected_duplicates, 0);
452 1 : BOOST_CHECK(tx.armed());
453 :
454 1 : tx.set_accepting(true);
455 1 : registry.callbacks.at("stream-a")(FakePayload{});
456 1 : BOOST_CHECK_EQUAL(tx.snapshot_stats().at("stream-a").sent, 1);
457 1 : }
458 :
459 2 : BOOST_AUTO_TEST_CASE(TransmitterObserverSeesPostTxTruthOnSuccessOnly)
460 : {
461 1 : test_transmitter_t tx;
462 1 : tx.add_stream("stream-a", 0);
463 1 : tx.configure(wibeth::PacketConfig{}, 0, FakeTxBackend{});
464 1 : tx.set_accepting(true);
465 :
466 1 : int observer_calls = 0;
467 1 : std::vector<std::uint8_t> observed;
468 2 : tx.set_post_tx_observer([&](const std::uint8_t* bytes, std::size_t len) {
469 1 : ++observer_calls;
470 1 : observed.assign(bytes, bytes + len);
471 1 : });
472 :
473 1 : tx.transmit("stream-a", FakePayload{});
474 1 : BOOST_REQUIRE_EQUAL(observer_calls, 1);
475 1 : BOOST_REQUIRE_EQUAL(observed.size(), wibeth::kEthernetPacketBytes);
476 1 : BOOST_CHECK(std::memcmp(observed.data(), tx.backend().storage.data(), wibeth::kEthernetPacketBytes) == 0);
477 :
478 : // Failed transmits are not observed.
479 1 : tx.backend().fail_transmit = true;
480 1 : tx.transmit("stream-a", FakePayload{});
481 1 : BOOST_CHECK_EQUAL(observer_calls, 1);
482 1 : BOOST_CHECK_EQUAL(tx.snapshot_stats().at("stream-a").tx_failures, 1);
483 1 : }
484 :
485 : BOOST_AUTO_TEST_SUITE_END()
|
