Skip to content

Commit 890dbdd

Browse files
TrentHoulistonTrentHouliston
committed
refactor: improve code quality and add packet processing tests
- Convert iovec C-arrays to std::array<iovec, N> (4 sites in NUClearNet.cpp, 1 in header) - Replace for(;;){break} in read_socket with while loop - Split process_packet switch/case into if-else chain dispatching to per-type methods: process_announce_packet, process_leave_packet, process_connect_packet, process_data_packet, process_ack_packet - Make per-type methods public for testability - Add comprehensive behavioral tests (ProcessPacket.cpp) covering: validate_header edge cases, packet dispatch, connect handshake, data delivery/rejection/dedup/reassembly, ACK handling, send paths
1 parent 89e3c38 commit 890dbdd

3 files changed

Lines changed: 669 additions & 163 deletions

File tree

‎src/nuclearnet/NUClearNet.cpp‎

Lines changed: 168 additions & 156 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
#include "NUClearNet.hpp"
2424

2525
#include <algorithm>
26+
#include <array>
2627
#include <cstring>
2728
#include <stdexcept>
2829
#include <system_error>
@@ -258,13 +259,12 @@ namespace network {
258259
header.flags = req.flags;
259260
header.hash = req.hash;
260261

261-
iovec iov[2];
262-
iov[0].iov_base = reinterpret_cast<void*>(&header);
263-
iov[0].iov_len = sizeof(DataPacket) - 1;
264-
iov[1].iov_base = const_cast<void*>(static_cast<const void*>(req.data.data()));
265-
iov[1].iov_len = req.data.size();
262+
std::array<iovec, 2> iov{{
263+
{reinterpret_cast<void*>(&header), sizeof(DataPacket) - 1}, // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
264+
{const_cast<void*>(static_cast<const void*>(req.data.data())), req.data.size()}, // NOLINT(cppcoreguidelines-pro-type-const-cast)
265+
}};
266266

267-
send_iov(data_fd, req.target, iov, 2);
267+
send_iov(data_fd, req.target, iov.data(), static_cast<int>(iov.size()));
268268
}
269269

270270
// Clean up expired fragment assemblies
@@ -312,13 +312,12 @@ namespace network {
312312
std::size_t offset = static_cast<std::size_t>(i) * packet_mtu;
313313
std::size_t frag_len = std::min(static_cast<std::size_t>(packet_mtu), length - offset);
314314

315-
iovec iov[2];
316-
iov[0].iov_base = reinterpret_cast<void*>(&header);
317-
iov[0].iov_len = sizeof(DataPacket) - 1;
318-
iov[1].iov_base = const_cast<void*>(static_cast<const void*>(payload + offset));
319-
iov[1].iov_len = frag_len;
315+
std::array<iovec, 2> iov{{
316+
{reinterpret_cast<void*>(&header), sizeof(DataPacket) - 1}, // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
317+
{const_cast<void*>(static_cast<const void*>(payload + offset)), frag_len}, // NOLINT(cppcoreguidelines-pro-type-const-cast)
318+
}};
320319

321-
send_iov(data_fd, announce_target, iov, 2);
320+
send_iov(data_fd, announce_target, iov.data(), static_cast<int>(iov.size()));
322321
}
323322
return;
324323
}
@@ -364,13 +363,12 @@ namespace network {
364363
std::size_t offset = static_cast<std::size_t>(i) * packet_mtu;
365364
std::size_t frag_len = std::min(static_cast<std::size_t>(packet_mtu), length - offset);
366365

367-
iovec iov[2];
368-
iov[0].iov_base = reinterpret_cast<void*>(&header);
369-
iov[0].iov_len = sizeof(DataPacket) - 1;
370-
iov[1].iov_base = const_cast<void*>(static_cast<const void*>(payload + offset));
371-
iov[1].iov_len = frag_len;
366+
std::array<iovec, 2> iov{{
367+
{reinterpret_cast<void*>(&header), sizeof(DataPacket) - 1}, // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
368+
{const_cast<void*>(static_cast<const void*>(payload + offset)), frag_len}, // NOLINT(cppcoreguidelines-pro-type-const-cast)
369+
}};
372370

373-
send_iov(data_fd, tgt, iov, 2);
371+
send_iov(data_fd, tgt, iov.data(), static_cast<int>(iov.size()));
374372
}
375373

376374
// If reliable, track for retransmission (single copy stored internally)
@@ -435,20 +433,24 @@ namespace network {
435433
// Drain all pending datagrams from the socket
436434
// The data socket is non-blocking, so recvfrom returns -1/EWOULDBLOCK when empty
437435
// For the announce socket, MSG_DONTWAIT provides the same behavior on POSIX
438-
for (;;) {
439-
socklen_t source_len = sizeof(source.storage);
440-
ssize_t received = ::recvfrom(fd,
441-
reinterpret_cast<char*>(buffer),
442-
sizeof(buffer),
443-
MSG_DONTWAIT,
444-
&source.sock,
445-
&source_len);
446-
447-
if (received <= 0) {
448-
break;
449-
}
450-
436+
socklen_t source_len = sizeof(source.storage);
437+
ssize_t received = ::recvfrom(fd,
438+
reinterpret_cast<char*>(buffer),
439+
sizeof(buffer),
440+
MSG_DONTWAIT,
441+
&source.sock,
442+
&source_len);
443+
444+
while (received > 0) {
451445
process_packet(source, buffer, static_cast<std::size_t>(received));
446+
447+
source_len = sizeof(source.storage);
448+
received = ::recvfrom(fd,
449+
reinterpret_cast<char*>(buffer),
450+
sizeof(buffer),
451+
MSG_DONTWAIT,
452+
&source.sock,
453+
&source_len);
452454
}
453455
}
454456

@@ -457,158 +459,168 @@ namespace network {
457459
return;
458460
}
459461

460-
const auto* header = reinterpret_cast<const PacketHeader*>(data);
462+
const auto* header = reinterpret_cast<const PacketHeader*>(data); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
461463

462464
// Touch the peer to reset timeout
463465
discovery->touch_peer(source);
464466

465-
switch (header->type) {
466-
case ANNOUNCE: {
467-
auto announce_result = discovery->process_announce(source, data, length);
468-
469-
if (announce_result.is_new) {
470-
// Force an immediate announce to the multicast/broadcast group
471-
// so the new peer hears us on the announce channel (confirms our_d→their_a)
472-
announce();
473-
last_announce = std::chrono::steady_clock::now();
474-
}
467+
if (header->type == ANNOUNCE) {
468+
process_announce_packet(source, data, length);
469+
}
470+
else if (header->type == LEAVE) {
471+
process_leave_packet(source);
472+
}
473+
else if (header->type == CONNECT) {
474+
process_connect_packet(source, data, length);
475+
}
476+
else if (header->type == DATA) {
477+
process_data_packet(source, data, length);
478+
}
479+
else if (header->type == ACK) {
480+
process_ack_packet(source, data, length);
481+
}
482+
}
475483

476-
// Send CONNECT packet if the handshake needs it (initial SYN or retransmit)
477-
if (announce_result.response_flags != 0) {
478-
auto pkt = Discovery::build_connect_packet(announce_result.response_flags);
479-
send_buf(data_fd, source, pkt.data(), pkt.size());
484+
void NUClearNet::process_announce_packet(const sock_t& source, const uint8_t* data, std::size_t length) {
485+
auto announce_result = discovery->process_announce(source, data, length);
480486

481-
// Mark SYN_SENT if we're sending a SYN (only advances from IDLE)
482-
if ((announce_result.response_flags & SYN) != 0) {
483-
discovery->mark_syn_sent(source);
484-
}
485-
}
486-
} break;
487+
if (announce_result.is_new) {
488+
// Force an immediate announce to the multicast/broadcast group
489+
// so the new peer hears us on the announce channel (confirms our_d→their_a)
490+
announce();
491+
last_announce = std::chrono::steady_clock::now();
492+
}
487493

488-
case LEAVE: {
489-
discovery->process_leave(source);
490-
} break;
494+
// Send CONNECT packet if the handshake needs it (initial SYN or retransmit)
495+
if (announce_result.response_flags != 0) {
496+
auto pkt = Discovery::build_connect_packet(announce_result.response_flags);
497+
send_buf(data_fd, source, pkt.data(), pkt.size());
491498

492-
case CONNECT: {
493-
if (length < sizeof(ConnectPacket)) {
494-
return;
495-
}
496-
const auto* pkt = reinterpret_cast<const ConnectPacket*>(data);
497-
auto result = discovery->process_connect(source, pkt->flags);
499+
// Mark SYN_SENT if we're sending a SYN (only advances from IDLE)
500+
if ((announce_result.response_flags & SYN) != 0) {
501+
discovery->mark_syn_sent(source);
502+
}
503+
}
504+
}
498505

499-
// Send response if the state machine requires one
500-
if (result.response_flags != 0) {
501-
auto response = Discovery::build_connect_packet(result.response_flags);
502-
send_buf(data_fd, source, response.data(), response.size());
503-
}
504-
} break;
506+
void NUClearNet::process_leave_packet(const sock_t& source) {
507+
discovery->process_leave(source);
508+
}
505509

506-
case DATA: {
507-
// Only accept data from connected peers
508-
if (!discovery->is_connected(source)) {
509-
return;
510-
}
510+
void NUClearNet::process_connect_packet(const sock_t& source, const uint8_t* data, std::size_t length) {
511+
if (length < sizeof(ConnectPacket)) {
512+
return;
513+
}
514+
const auto* pkt = reinterpret_cast<const ConnectPacket*>(data); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
515+
auto result = discovery->process_connect(source, pkt->flags);
511516

512-
if (length < sizeof(DataPacket)) {
513-
return;
514-
}
517+
// Send response if the state machine requires one
518+
if (result.response_flags != 0) {
519+
auto response = Discovery::build_connect_packet(result.response_flags);
520+
send_buf(data_fd, source, response.data(), response.size());
521+
}
522+
}
515523

516-
const auto* pkt = reinterpret_cast<const DataPacket*>(data);
524+
void NUClearNet::process_data_packet(const sock_t& source, const uint8_t* data, std::size_t length) {
525+
// Only accept data from connected peers
526+
if (!discovery->is_connected(source)) {
527+
return;
528+
}
517529

518-
// Drop messages we are not subscribed to (relevant for multicast broadcast data)
519-
if (!routing.is_locally_subscribed(pkt->hash)) {
520-
return;
521-
}
530+
if (length < sizeof(DataPacket)) {
531+
return;
532+
}
522533

523-
// Check for duplicates (at the packet group level)
524-
auto& dedup = deduplicators[source];
534+
const auto* pkt = reinterpret_cast<const DataPacket*>(data); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
525535

526-
if (dedup.is_duplicate(pkt->packet_id)) {
527-
// Already processed this packet group — send ACK if reliable
528-
if ((pkt->flags & RELIABLE) != 0) {
529-
std::vector<bool> all_received(pkt->packet_count, true);
530-
auto ack = Reliability::build_ack_packet(pkt->packet_id, pkt->packet_count, all_received);
531-
send_buf(data_fd, source, ack.data(), ack.size());
532-
}
533-
return;
534-
}
536+
// Drop messages we are not subscribed to (relevant for multicast broadcast data)
537+
if (!routing.is_locally_subscribed(pkt->hash)) {
538+
return;
539+
}
535540

536-
// Extract fragment data
537-
const uint8_t* frag_data = data + sizeof(DataPacket) - 1;
538-
std::size_t frag_length = length - (sizeof(DataPacket) - 1);
541+
// Check for duplicates (at the packet group level)
542+
auto& dedup = deduplicators[source];
539543

540-
// Use a hash of the full source address as the source key for fragmentation
541-
// This ensures IPv6 addresses are properly distinguished
542-
uint64_t source_key = 0;
543-
const auto* bytes = reinterpret_cast<const uint8_t*>(&source.storage);
544-
for (std::size_t i = 0; i < sizeof(source.storage); ++i) {
545-
source_key ^= static_cast<uint64_t>(bytes[i]) << ((i % 8) * 8);
546-
}
544+
if (dedup.is_duplicate(pkt->packet_id)) {
545+
// Already processed this packet group — send ACK if reliable
546+
if ((pkt->flags & RELIABLE) != 0) {
547+
std::vector<bool> all_received(pkt->packet_count, true);
548+
auto ack = Reliability::build_ack_packet(pkt->packet_id, pkt->packet_count, all_received);
549+
send_buf(data_fd, source, ack.data(), ack.size());
550+
}
551+
return;
552+
}
547553

548-
// Submit to fragmentation
549-
Fragmentation::AssembledPacket assembled;
550-
bool has_assembled = fragmentation->submit_fragment(source_key,
551-
pkt->packet_id,
552-
pkt->packet_no,
553-
pkt->packet_count,
554-
pkt->hash,
555-
pkt->flags,
556-
frag_data,
557-
frag_length,
558-
assembled);
559-
560-
// Send ACK for reliable packets
561-
if ((pkt->flags & RELIABLE) != 0) {
562-
// Build a partial ACK (we'd need to track received fragments per assembly)
563-
// For now, send an ACK indicating we have this fragment
564-
std::vector<bool> received(pkt->packet_count, false);
565-
received[pkt->packet_no] = true;
566-
auto ack = Reliability::build_ack_packet(pkt->packet_id, pkt->packet_count, received);
567-
send_buf(data_fd, source, ack.data(), ack.size());
568-
}
554+
// Extract fragment data
555+
const uint8_t* frag_data = data + sizeof(DataPacket) - 1;
556+
std::size_t frag_length = length - (sizeof(DataPacket) - 1);
569557

570-
// If we have a complete message, deliver it
571-
if (has_assembled) {
572-
dedup.add_packet(assembled.packet_id);
558+
// Use a hash of the full source address as the source key for fragmentation
559+
// This ensures IPv6 addresses are properly distinguished
560+
uint64_t source_key = 0;
561+
const auto* bytes = reinterpret_cast<const uint8_t*>(&source.storage); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
562+
for (std::size_t i = 0; i < sizeof(source.storage); ++i) {
563+
source_key ^= static_cast<uint64_t>(bytes[i]) << ((i % 8) * 8);
564+
}
573565

574-
if (packet_callback) {
575-
// Look up peer name
576-
std::string peer_name;
577-
const auto* peer = discovery->get_peer(source);
578-
if (peer != nullptr) {
579-
peer_name = peer->name;
580-
}
566+
// Submit to fragmentation
567+
Fragmentation::AssembledPacket assembled;
568+
bool has_assembled = fragmentation->submit_fragment(source_key,
569+
pkt->packet_id,
570+
pkt->packet_no,
571+
pkt->packet_count,
572+
pkt->hash,
573+
pkt->flags,
574+
frag_data,
575+
frag_length,
576+
assembled);
577+
578+
// Send ACK for reliable packets
579+
if ((pkt->flags & RELIABLE) != 0) {
580+
std::vector<bool> received(pkt->packet_count, false);
581+
received[pkt->packet_no] = true;
582+
auto ack = Reliability::build_ack_packet(pkt->packet_id, pkt->packet_count, received);
583+
send_buf(data_fd, source, ack.data(), ack.size());
584+
}
581585

582-
bool reliable = (assembled.flags & RELIABLE) != 0;
583-
packet_callback(source, peer_name, assembled.hash, reliable, std::move(assembled.payload));
584-
}
586+
// If we have a complete message, deliver it
587+
if (has_assembled) {
588+
dedup.add_packet(assembled.packet_id);
585589

586-
// Send full ACK for reliable
587-
if ((assembled.flags & RELIABLE) != 0) {
588-
std::vector<bool> all_received(pkt->packet_count, true);
589-
auto ack = Reliability::build_ack_packet(pkt->packet_id, pkt->packet_count, all_received);
590-
send_buf(data_fd, source, ack.data(), ack.size());
591-
}
590+
if (packet_callback) {
591+
// Look up peer name
592+
std::string peer_name;
593+
const auto* peer = discovery->get_peer(source);
594+
if (peer != nullptr) {
595+
peer_name = peer->name;
592596
}
593-
} break;
594597

595-
case ACK: {
596-
// Only accept ACKs from connected peers
597-
if (!discovery->is_connected(source)) {
598-
return;
599-
}
598+
bool reliable_flag = (assembled.flags & RELIABLE) != 0;
599+
packet_callback(source, peer_name, assembled.hash, reliable_flag, std::move(assembled.payload));
600+
}
600601

601-
if (length < sizeof(ACKPacket)) {
602-
return;
603-
}
604-
const auto* pkt = reinterpret_cast<const ACKPacket*>(data);
605-
const uint8_t* bitset = data + sizeof(ACKPacket) - 1;
606-
std::size_t bitset_size = length - (sizeof(ACKPacket) - 1);
607-
reliability->process_ack(source, pkt->packet_id, pkt->packet_count, bitset, bitset_size);
608-
} break;
602+
// Send full ACK for reliable
603+
if ((assembled.flags & RELIABLE) != 0) {
604+
std::vector<bool> all_received(pkt->packet_count, true);
605+
auto ack = Reliability::build_ack_packet(pkt->packet_id, pkt->packet_count, all_received);
606+
send_buf(data_fd, source, ack.data(), ack.size());
607+
}
608+
}
609+
}
610+
611+
void NUClearNet::process_ack_packet(const sock_t& source, const uint8_t* data, std::size_t length) {
612+
// Only accept ACKs from connected peers
613+
if (!discovery->is_connected(source)) {
614+
return;
615+
}
609616

610-
default: break;
617+
if (length < sizeof(ACKPacket)) {
618+
return;
611619
}
620+
const auto* pkt = reinterpret_cast<const ACKPacket*>(data); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
621+
const uint8_t* bitset = data + sizeof(ACKPacket) - 1;
622+
std::size_t bitset_size = length - (sizeof(ACKPacket) - 1);
623+
reliability->process_ack(source, pkt->packet_id, pkt->packet_count, bitset, bitset_size);
612624
}
613625

614626
void NUClearNet::send_iov(fd_t fd, const sock_t& target, const iovec* iov, int iovcnt) {

0 commit comments

Comments
 (0)