Skip to content

Commit 9e21daa

Browse files
pavel-kirienkopavel-kirienko
committed
fix tests
1 parent ab57867 commit 9e21daa

File tree

7 files changed

+277
-218
lines changed

7 files changed

+277
-218
lines changed

‎AGENTS.md‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
1-
# LibUDPard instructions for agents
1+
# LibUDPard instructions for AI agents
22

3-
Please read README.md for general information about LibUDPard.
4-
The library source files are just two: `libudpard/udpard.c` and `libudpard/udpard.h`.
3+
Please read `README.md` for general information about LibUDPard.
54

65
Keep the code and comments very brief. Be sure every significant code block is preceded with a brief comment.
76

‎README.md‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,21 @@ next-generation intelligent vehicles: manned and unmanned aircraft, spacecraft,
2222

2323
- Zero-copy RX pipeline -- payload is moved from the NIC driver all the way to the application without copying.
2424
- Support for redundant network interfaces with seamless interface aggregation and zero fail-over delay.
25-
- Robust message reassembler tolerant to highly distorted datagram streams (out-of-order, duplication, distinct MTU).
26-
- Message ordering recovery for ordering-sensitive applications (e.g., state estimators, control loops).
25+
- Robust message reassembler supporting highly distorted datagram streams:
26+
out-of-order fragments, message ordering recovery, fragment/message deduplication, interleaving, variable MTU, ...
27+
- Robust message ordering recovery for ordering-sensitive applications (e.g., state estimators, control loops)
28+
with well-defined deterministic recovery in the event of lost messages.
2729
- Packet loss mitigation via:
28-
- repetition-coding FEC (transparent to the application);
2930
- redundant interfaces (packet lost on one interface may be received on another, transparent to the application);
30-
- positive acknowledgment with retransmission (retransmission not handled by the library).
31+
- reliable topics (retransmit until acknowledged; callback notifications for successful/failed deliveries).
32+
- Single-copy TX pipeline with fragment deduplication across multiple interfaces and reference counting.
3133
- Heap not required; the library can be used with fixed-size block pool allocators.
3234
- Detailed time complexity and memory requirement models for the benefit of real-time high-integrity applications.
33-
- Runs on any 8/16/32/64-bit platform and extremely resource-constrained baremetal environments with ~100K ROM/RAM.
35+
- Runs anywhere out of the box, including extremely resource-constrained baremetal environments with ~100K ROM/RAM.
36+
No porting required.
3437
- MISRA C compliance (reach out to <https://forum.opencyphal.org>).
35-
- Full implementation in a single C file with less than 2k lines of straightforward code!
38+
- Full implementation in a single C file with only ~2k lines of straightforward C99!
39+
- Extensive test coverage.
3640

3741
## Usage
3842

@@ -72,7 +76,9 @@ standards-compliant C99 compiler is available.
7276

7377
### v3.0
7478

75-
WIP --- adding support for Cyphal v1.1.
79+
The library has been redesigned from scratch to support Cyphal v1.1, named topics, and reliable transfers.
80+
No porting guide is provided since the changes are too significant;
81+
please refer to the new API docs in `libudpard/udpard.h`.
7682

7783
### v2.0
7884

‎libudpard/udpard.c‎

Lines changed: 11 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -921,6 +921,7 @@ static void tx_receive_ack(udpard_rx_t* const rx, const uint64_t topic_hash, con
921921
}
922922

923923
/// Generate an ack transfer for the specified remote transfer.
924+
/// Do nothing if an ack for the same transfer is already enqueued with equal or better endpoint coverage.
924925
static void tx_send_ack(udpard_rx_t* const rx,
925926
const udpard_us_t now,
926927
const udpard_prio_t priority,
@@ -943,7 +944,7 @@ static void tx_send_ack(udpard_rx_t* const rx,
943944
return; // Can we get an ack? We have ack at home!
944945
}
945946
if (prior != NULL) {
946-
tx_transfer_free(tx, prior); // avoid redundant acks for the same transfer
947+
tx_transfer_free(tx, prior); // avoid redundant acks for the same transfer -- replace with better one
947948
}
948949

949950
// Serialize the ACK payload.
@@ -958,24 +959,15 @@ static void tx_send_ack(udpard_rx_t* const rx,
958959

959960
// Enqueue the transfer.
960961
const udpard_bytes_t payload = { .size = UDPARD_P2P_HEADER_BYTES, .data = header };
961-
const meta_t meta = {
962-
.priority = priority,
963-
.flag_ack = false,
964-
.transfer_payload_size = (uint32_t)payload.size,
965-
.transfer_id = tx->p2p_transfer_id++,
966-
.sender_uid = tx->local_uid,
967-
.topic_hash = remote.uid, // this is a P2P transfer
968-
};
969-
tx_transfer_t* tr = NULL;
970-
const uint32_t count = tx_push(tx, //
971-
now,
972-
now + ACK_TX_DEADLINE,
973-
meta,
974-
remote.endpoints,
975-
payload,
976-
NULL,
977-
NULL,
978-
&tr);
962+
const meta_t meta = { .priority = priority,
963+
.flag_ack = false,
964+
.transfer_payload_size = (uint32_t)payload.size,
965+
.transfer_id = tx->p2p_transfer_id++,
966+
.sender_uid = tx->local_uid,
967+
.topic_hash = remote.uid };
968+
tx_transfer_t* tr = NULL;
969+
const uint32_t count =
970+
tx_push(tx, now, now + ACK_TX_DEADLINE, meta, remote.endpoints, payload, NULL, NULL, &tr);
979971
UDPARD_ASSERT(count <= 1);
980972
if (count == 1) { // ack is always a single-frame transfer, so we get either 0 or 1
981973
UDPARD_ASSERT(tr != NULL);

‎libudpard/udpard.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -716,6 +716,8 @@ struct udpard_rx_port_p2p_t
716716

717717
/// The RX instance holds no resources and can be destroyed at any time by simply freeing all its ports first
718718
/// using udpard_rx_port_free(), then discarding the instance itself. The self pointer must not be NULL.
719+
/// The TX instance must be initialized beforehand, unless the application wants to only listen,
720+
/// in which case it may be NULL.
719721
void udpard_rx_new(udpard_rx_t* const self, udpard_tx_t* const tx);
720722

721723
/// Must be invoked at least every few milliseconds (more often is fine) to purge timed-out sessions and eject

‎tests/src/test_e2e_edge.cpp‎

Lines changed: 83 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -15,30 +15,58 @@ namespace {
1515

1616
void on_message(udpard_rx_t* rx, udpard_rx_port_t* port, udpard_rx_transfer_t transfer);
1717
void on_collision(udpard_rx_t* rx, udpard_rx_port_t* port, udpard_remote_t remote);
18-
constexpr udpard_rx_port_vtable_t callbacks{ &on_message, &on_collision };
18+
constexpr udpard_rx_port_vtable_t callbacks{ .on_message = &on_message, .on_collision = &on_collision };
19+
20+
struct CapturedFrame
21+
{
22+
udpard_bytes_mut_t datagram;
23+
uint_fast8_t iface_index;
24+
};
25+
26+
void tx_refcount_free(void* const user, const size_t size, void* const payload)
27+
{
28+
(void)user;
29+
udpard_tx_refcount_dec(udpard_bytes_t{ .size = size, .data = payload });
30+
}
31+
32+
bool capture_tx_frame(udpard_tx_t* const tx, const udpard_tx_ejection_t ejection)
33+
{
34+
auto* frames = static_cast<std::vector<CapturedFrame>*>(tx->user);
35+
if (frames == nullptr) {
36+
return false;
37+
}
38+
udpard_tx_refcount_inc(ejection.datagram);
39+
void* const data = const_cast<void*>(ejection.datagram.data); // NOLINT(cppcoreguidelines-pro-type-const-cast)
40+
frames->push_back(CapturedFrame{ .datagram = { .size = ejection.datagram.size, .data = data },
41+
.iface_index = ejection.iface_index });
42+
return true;
43+
}
44+
45+
constexpr udpard_tx_vtable_t tx_vtable{ .eject = &capture_tx_frame };
1946

2047
struct Context
2148
{
2249
std::vector<uint64_t> ids;
2350
size_t collisions = 0;
2451
uint64_t expected_uid = 0;
25-
udpard_udpip_ep_t source = {};
52+
udpard_udpip_ep_t source{};
2653
};
2754

2855
struct Fixture
2956
{
30-
instrumented_allocator_t tx_alloc_frag{};
31-
instrumented_allocator_t tx_alloc_payload{};
32-
instrumented_allocator_t rx_alloc_frag{};
33-
instrumented_allocator_t rx_alloc_session{};
34-
udpard_tx_t tx{};
35-
udpard_rx_t rx{};
36-
udpard_rx_port_t port{};
37-
udpard_mem_deleter_t tx_payload_deleter{};
38-
Context ctx{};
39-
udpard_udpip_ep_t dest{};
40-
udpard_udpip_ep_t source{};
41-
uint64_t topic_hash{ 0x90AB12CD34EF5678ULL };
57+
instrumented_allocator_t tx_alloc_transfer{};
58+
instrumented_allocator_t tx_alloc_payload{};
59+
instrumented_allocator_t rx_alloc_frag{};
60+
instrumented_allocator_t rx_alloc_session{};
61+
udpard_tx_t tx{};
62+
udpard_rx_t rx{};
63+
udpard_rx_port_t port{};
64+
udpard_mem_deleter_t tx_payload_deleter{};
65+
std::vector<CapturedFrame> frames;
66+
Context ctx{};
67+
udpard_udpip_ep_t dest{};
68+
udpard_udpip_ep_t source{};
69+
uint64_t topic_hash{ 0x90AB12CD34EF5678ULL };
4270

4371
Fixture(const Fixture&) = delete;
4472
Fixture& operator=(const Fixture&) = delete;
@@ -47,21 +75,24 @@ struct Fixture
4775

4876
explicit Fixture(const udpard_us_t reordering_window)
4977
{
50-
instrumented_allocator_new(&tx_alloc_frag);
78+
instrumented_allocator_new(&tx_alloc_transfer);
5179
instrumented_allocator_new(&tx_alloc_payload);
5280
instrumented_allocator_new(&rx_alloc_frag);
5381
instrumented_allocator_new(&rx_alloc_session);
54-
const udpard_tx_mem_resources_t tx_mem{ .fragment = instrumented_allocator_make_resource(&tx_alloc_frag),
55-
.payload = instrumented_allocator_make_resource(&tx_alloc_payload) };
82+
udpard_tx_mem_resources_t tx_mem{};
83+
tx_mem.transfer = instrumented_allocator_make_resource(&tx_alloc_transfer);
84+
for (auto& res : tx_mem.payload) {
85+
res = instrumented_allocator_make_resource(&tx_alloc_payload);
86+
}
5687
const udpard_rx_mem_resources_t rx_mem{ .session = instrumented_allocator_make_resource(&rx_alloc_session),
5788
.fragment = instrumented_allocator_make_resource(&rx_alloc_frag) };
58-
tx_payload_deleter = instrumented_allocator_make_deleter(&tx_alloc_payload);
89+
tx_payload_deleter = udpard_mem_deleter_t{ .user = nullptr, .free = &tx_refcount_free };
5990
source = { .ip = 0x0A000001U, .port = 7501U };
6091
dest = udpard_make_subject_endpoint(222U);
6192

62-
TEST_ASSERT_TRUE(udpard_tx_new(&tx, 0x0A0B0C0D0E0F1011ULL, 16, tx_mem));
63-
std::array<udpard_tx_t*, UDPARD_NETWORK_INTERFACE_COUNT_MAX> rx_tx{};
64-
udpard_rx_new(&rx, rx_tx.data(), 0);
93+
TEST_ASSERT_TRUE(udpard_tx_new(&tx, 0x0A0B0C0D0E0F1011ULL, 42U, 16, tx_mem, &tx_vtable));
94+
tx.user = &frames;
95+
udpard_rx_new(&rx, nullptr);
6596
ctx.expected_uid = tx.local_uid;
6697
ctx.source = source;
6798
rx.user = &ctx;
@@ -71,36 +102,48 @@ struct Fixture
71102
~Fixture()
72103
{
73104
udpard_rx_port_free(&rx, &port);
105+
udpard_tx_free(&tx);
74106
TEST_ASSERT_EQUAL_size_t(0, rx_alloc_frag.allocated_fragments);
75107
TEST_ASSERT_EQUAL_size_t(0, rx_alloc_session.allocated_fragments);
76-
TEST_ASSERT_EQUAL_size_t(0, tx_alloc_frag.allocated_fragments);
108+
TEST_ASSERT_EQUAL_size_t(0, tx_alloc_transfer.allocated_fragments);
77109
TEST_ASSERT_EQUAL_size_t(0, tx_alloc_payload.allocated_fragments);
78110
instrumented_allocator_reset(&rx_alloc_frag);
79111
instrumented_allocator_reset(&rx_alloc_session);
80-
instrumented_allocator_reset(&tx_alloc_frag);
112+
instrumented_allocator_reset(&tx_alloc_transfer);
81113
instrumented_allocator_reset(&tx_alloc_payload);
82114
}
83115

84116
void push_single(const udpard_us_t ts, const uint64_t transfer_id)
85117
{
118+
frames.clear();
86119
std::array<uint8_t, 8> payload_buf{};
87120
for (size_t i = 0; i < payload_buf.size(); i++) {
88121
payload_buf[i] = static_cast<uint8_t>(transfer_id >> (i * 8U));
89122
}
90123
const udpard_bytes_t payload{ .size = payload_buf.size(), .data = payload_buf.data() };
91-
const udpard_us_t deadline = ts + 1000000;
92-
const uint_fast8_t iface_index = 0;
93-
TEST_ASSERT_GREATER_THAN_UINT32(
94-
0U,
95-
udpard_tx_push(&tx, ts, deadline, udpard_prio_slow, topic_hash, dest, transfer_id, payload, false, nullptr));
96-
udpard_tx_item_t* const item = udpard_tx_peek(&tx, ts);
97-
TEST_ASSERT_NOT_NULL(item);
98-
udpard_tx_pop(&tx, item);
99-
TEST_ASSERT_TRUE(
100-
udpard_rx_port_push(&rx, &port, ts, source, item->datagram_payload, tx_payload_deleter, iface_index));
101-
item->datagram_payload.data = nullptr;
102-
item->datagram_payload.size = 0;
103-
udpard_tx_free(tx.memory, item);
124+
const udpard_us_t deadline = ts + 1000000;
125+
for (auto& mtu_value : tx.mtu) {
126+
mtu_value = UDPARD_MTU_DEFAULT;
127+
}
128+
std::array<udpard_udpip_ep_t, UDPARD_IFACE_COUNT_MAX> dest_per_iface{};
129+
dest_per_iface.fill(udpard_udpip_ep_t{});
130+
dest_per_iface[0] = dest;
131+
TEST_ASSERT_GREATER_THAN_UINT32(0U,
132+
udpard_tx_push(&tx,
133+
ts,
134+
deadline,
135+
udpard_prio_slow,
136+
topic_hash,
137+
dest_per_iface.data(),
138+
transfer_id,
139+
payload,
140+
nullptr,
141+
nullptr));
142+
udpard_tx_poll(&tx, ts, UDPARD_IFACE_MASK_ALL);
143+
TEST_ASSERT_GREATER_THAN_UINT32(0U, static_cast<uint32_t>(frames.size()));
144+
for (const auto& [datagram, iface_index] : frames) {
145+
TEST_ASSERT_TRUE(udpard_rx_port_push(&rx, &port, ts, source, datagram, tx_payload_deleter, iface_index));
146+
}
104147
}
105148
};
106149

@@ -127,15 +170,15 @@ void test_udpard_rx_unordered_duplicates()
127170
Fixture fix{ UDPARD_RX_REORDERING_WINDOW_UNORDERED };
128171
udpard_us_t now = 0;
129172

130-
const std::array<uint64_t, 6> ids{ 100, 20000, 10100, 5000, 20000, 100 };
173+
constexpr std::array<uint64_t, 6> ids{ 100, 20000, 10100, 5000, 20000, 100 };
131174
for (const auto id : ids) {
132175
fix.push_single(now, id);
133176
udpard_rx_poll(&fix.rx, now);
134177
now++;
135178
}
136179
udpard_rx_poll(&fix.rx, now + 100);
137180

138-
const std::array<uint64_t, 4> expected{ 100, 20000, 10100, 5000 };
181+
constexpr std::array<uint64_t, 4> expected{ 100, 20000, 10100, 5000 };
139182
TEST_ASSERT_EQUAL_size_t(expected.size(), fix.ctx.ids.size());
140183
for (size_t i = 0; i < expected.size(); i++) {
141184
TEST_ASSERT_EQUAL_UINT64(expected[i], fix.ctx.ids[i]);
@@ -176,7 +219,7 @@ void test_udpard_rx_ordered_out_of_order()
176219
// Allow the window to expire so the remaining interned transfers eject.
177220
udpard_rx_poll(&fix.rx, now + 70);
178221

179-
const std::array<uint64_t, 5> expected{ 100, 200, 300, 10100, 10200 };
222+
constexpr std::array<uint64_t, 5> expected{ 100, 200, 300, 10100, 10200 };
180223
TEST_ASSERT_EQUAL_size_t(expected.size(), fix.ctx.ids.size());
181224
for (size_t i = 0; i < expected.size(); i++) {
182225
TEST_ASSERT_EQUAL_UINT64(expected[i], fix.ctx.ids[i]);
@@ -211,7 +254,7 @@ void test_udpard_rx_ordered_head_advanced_late()
211254
fix.push_single(++now, 310);
212255
udpard_rx_poll(&fix.rx, now);
213256

214-
const std::array<uint64_t, 5> expected{ 100, 200, 300, 420, 450 };
257+
constexpr std::array<uint64_t, 5> expected{ 100, 200, 300, 420, 450 };
215258
TEST_ASSERT_EQUAL_size_t(expected.size(), fix.ctx.ids.size());
216259
for (size_t i = 0; i < expected.size(); i++) {
217260
TEST_ASSERT_EQUAL_UINT64(expected[i], fix.ctx.ids[i]);

0 commit comments

Comments
 (0)