Skip to content

Commit 8ba629b

Browse files
lydia-at-amazonlydia-at-amazon
authored
Restart RX Session if out-of-order multiframe transfer received (#20)
During on-target testing, it was found that if a frame is received out of order during a multiframe transfer, then libudpard cannot receive any subsequent transfers for any port IDs because it reports an out of memory error. This is because the memory allocated for the out-of-order multiframe transfer is never freed. Since out-of-order multiframe transfers are not currently supported, the best course of action is drop the entire payload and require the user to resend it.
1 parent 89f87ab commit 8ba629b

File tree

3 files changed

+88
-43
lines changed

3 files changed

+88
-43
lines changed

‎libudpard/udpard.c‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -976,24 +976,29 @@ UDPARD_PRIVATE int8_t rxSessionUpdate(UdpardInstance* const ins,
976976
// multi-frame transfer
977977
if (!(frame->start_of_transfer && frame->end_of_transfer))
978978
{
979-
if (frame->end_of_transfer) {
980-
if (frame->frame_index !=
981-
(uint32_t) (1U << (uint32_t) UDPARD_END_OF_TRANSFER_OFFSET)
982-
+ rxs->last_udp_header_index + 1)
983-
{
984-
// Not the expected index
985-
int8_t out = -UDPARD_ERROR_OUT_OF_ORDER;
986-
// reset index to 0
987-
rxs->last_udp_header_index = 0U;
988-
return out;
989-
}
979+
if (frame->end_of_transfer)
980+
{
981+
uint32_t next_expected_frame_index = (1U << UDPARD_END_OF_TRANSFER_OFFSET) + rxs->last_udp_header_index + 1;
982+
if (frame->frame_index != next_expected_frame_index)
983+
{
984+
// Out of order multiframe packet received
985+
out = -UDPARD_ERROR_OUT_OF_ORDER;
986+
// Reset previous frame index to 0
990987
rxs->last_udp_header_index = 0U;
988+
rxSessionRestart(ins, rxs);
989+
return out;
990+
}
991+
rxs->last_udp_header_index = 0U;
991992
}
992993
else
993994
{
994-
if ((!frame->start_of_transfer && frame->frame_index != rxs->last_udp_header_index+1) || (frame->start_of_transfer && frame->frame_index != 1U))
995+
if ((!frame->start_of_transfer && frame->frame_index != rxs->last_udp_header_index + 1)
996+
|| (frame->start_of_transfer && frame->frame_index != 1U))
995997
{
996-
return -UDPARD_ERROR_OUT_OF_ORDER;
998+
// Out of order multiframe packet received
999+
out = -UDPARD_ERROR_OUT_OF_ORDER;
1000+
rxSessionRestart(ins, rxs);
1001+
return out;
9971002
}
9981003
rxs->last_udp_header_index = frame->frame_index;
9991004
}

‎tests/test_private_rx.cpp‎

Lines changed: 61 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@
77
#include "catch/catch.hpp"
88
#include <cstring>
99

10+
static const uint32_t CRC_INITIAL = 0xFFFFFFFFU;
11+
1012
TEST_CASE("rxTryParseFrame")
1113
{
1214
using exposed::RxFrameModel;
@@ -593,57 +595,99 @@ TEST_CASE("rxSessionUpdate")
593595
frame.payload_size = 2;
594596
frame.payload = reinterpret_cast<const uint8_t*>("\x09\x09");
595597
REQUIRE(-UDPARD_ERROR_OUT_OF_ORDER == update(0, 1'000'000, 16));
598+
// The session should be restarted if an out of order frame is received
599+
// and the entire transfer will be dropped. Verify that all variables have
600+
// been set back to defaults by rxSessionRestart.
601+
REQUIRE(rxs.total_payload_size == 0U);
602+
REQUIRE(rxs.payload_size == 0U);
603+
REQUIRE(rxs.payload == NULL);
604+
REQUIRE(rxs.calculated_crc == CRC_INITIAL);
596605
// Update another session using same frame, fail.
597606
REQUIRE(-UDPARD_ERROR_OUT_OF_ORDER == updateAnotherSession(1, 1'000'000, 16));
607+
REQUIRE(rxs_1.total_payload_size == 0U);
608+
REQUIRE(rxs_1.payload_size == 0U);
609+
REQUIRE(rxs_1.payload == NULL);
610+
REQUIRE(rxs_1.calculated_crc == CRC_INITIAL);
611+
612+
// Once you get an out-of-order frame in a multiframe transfer,
613+
// the entire payload needs to be resent, so start over.
614+
615+
// Multi-frame, first frame
616+
frame.timestamp_usec = 20'000'300;
617+
frame.transfer_id = 14;
618+
frame.start_of_transfer = true;
619+
frame.end_of_transfer = false;
620+
frame.payload_size = 7;
621+
frame.frame_index = 1;
622+
frame.payload = reinterpret_cast<const uint8_t*>("\x06\x06\x06\x06\x06\x06\x06");
623+
REQUIRE(0 == update(0, 1'000'000, 16));
624+
625+
REQUIRE(rxs.transfer_timestamp_usec == 20'000'300);
626+
REQUIRE(rxs.payload_size == 7);
627+
REQUIRE(0 == std::memcmp(rxs.payload, "\x06\x06\x06\x06\x06\x06\x06", 7));
628+
REQUIRE(rxs.calculated_crc == crc("\x06\x06\x06\x06\x06\x06\x06"));
629+
REQUIRE(rxs.transfer_id == 14U);
630+
REQUIRE(rxs.redundant_transport_index == 0);
631+
REQUIRE(ins.getAllocator().getNumAllocatedFragments() == 1);
632+
REQUIRE(ins.getAllocator().getTotalAllocatedAmount() == 16);
633+
634+
// Update another session using same frame.
635+
REQUIRE(0 == updateAnotherSession(1, 1'000'000, 16));
636+
REQUIRE(rxs_1.transfer_timestamp_usec == 20'000'300);
637+
REQUIRE(rxs_1.payload_size == 7);
638+
REQUIRE(0 == std::memcmp(rxs_1.payload, "\x06\x06\x06\x06\x06\x06\x06", 7));
639+
REQUIRE(rxs_1.calculated_crc == crc("\x06\x06\x06\x06\x06\x06\x06"));
640+
REQUIRE(rxs_1.transfer_id == 14U);
641+
REQUIRE(rxs_1.redundant_transport_index == 1);
642+
REQUIRE(ins_1.getAllocator().getNumAllocatedFragments() == 1);
643+
REQUIRE(ins_1.getAllocator().getTotalAllocatedAmount() == 16);
598644

599645
// Multi-frame, middle.
600-
frame.timestamp_usec = 20'000'200;
601646
frame.start_of_transfer = false;
602647
frame.end_of_transfer = false;
603648
frame.frame_index = 2;
604649
frame.payload_size = 7;
605650
frame.payload = reinterpret_cast<const uint8_t*>("\x07\x07\x07\x07\x07\x07\x07");
606651
REQUIRE(0 == update(0, 1'000'000, 16));
607-
REQUIRE(rxs.transfer_timestamp_usec == 20'000'100);
652+
REQUIRE(rxs.transfer_timestamp_usec == 20'000'300);
608653
REQUIRE(rxs.payload_size == 14);
609654
REQUIRE(0 == std::memcmp(rxs.payload, "\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07", 14));
610655
REQUIRE(rxs.calculated_crc == crc("\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07"));
611-
REQUIRE(rxs.transfer_id == 13U);
656+
REQUIRE(rxs.transfer_id == 14U);
612657
REQUIRE(rxs.redundant_transport_index == 0);
613658
REQUIRE(ins.getAllocator().getNumAllocatedFragments() == 1);
614659
REQUIRE(ins.getAllocator().getTotalAllocatedAmount() == 16);
615660

616661
// Update another session using same frame.
617662
REQUIRE(0 == updateAnotherSession(1, 1'000'000, 16));
618-
REQUIRE(rxs_1.transfer_timestamp_usec == 20'000'100);
663+
REQUIRE(rxs_1.transfer_timestamp_usec == 20'000'300);
619664
REQUIRE(rxs_1.payload_size == 14);
620665
REQUIRE(0 == std::memcmp(rxs_1.payload, "\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07", 14));
621666
REQUIRE(rxs_1.calculated_crc == crc("\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07"));
622-
REQUIRE(rxs_1.transfer_id == 13U);
667+
REQUIRE(rxs_1.transfer_id == 14U);
623668
REQUIRE(rxs_1.redundant_transport_index == 1);
624669
REQUIRE(ins_1.getAllocator().getNumAllocatedFragments() == 1);
625670
REQUIRE(ins_1.getAllocator().getTotalAllocatedAmount() == 16);
626671

627672
// Multi-frame, last.
628-
frame.timestamp_usec = 20'000'400;
629673
frame.start_of_transfer = false;
630674
frame.end_of_transfer = true;
631675
frame.frame_index = 3 + static_cast<uint32_t>(1U << static_cast<uint32_t>(31U));
632676
frame.payload_size = 8; // The payload is IMPLICITLY TRUNCATED, and the CRC IS STILL VALIDATED.
633677
frame.payload = reinterpret_cast<const uint8_t*>("\x09\x09\x09\x09\x32\x98\x04\x7B");
634678
REQUIRE(1 == update(0, 1'000'000, 16));
635-
REQUIRE(rxs.transfer_timestamp_usec == 20'000'100); // First frame.
679+
REQUIRE(rxs.transfer_timestamp_usec == 20'000'300); // The timestamp is the same as the first frame.
636680
REQUIRE(rxs.payload_size == 0);
637681
REQUIRE(rxs.payload == nullptr);
638682
REQUIRE(rxs.calculated_crc == 0xFFFFFFFFU);
639-
REQUIRE(rxs.transfer_id == 14U);
683+
REQUIRE(rxs.transfer_id == 15U);
640684
REQUIRE(rxs.redundant_transport_index == 0);
641-
REQUIRE(transfer.timestamp_usec == 20'000'100);
685+
REQUIRE(transfer.timestamp_usec == 20'000'300);
642686
REQUIRE(transfer.metadata.priority == UdpardPrioritySlow);
643687
REQUIRE(transfer.metadata.transfer_kind == UdpardTransferKindMessage);
644688
REQUIRE(transfer.metadata.port_id == 2'222);
645689
REQUIRE(transfer.metadata.remote_node_id == 55);
646-
REQUIRE(transfer.metadata.transfer_id == 13);
690+
REQUIRE(transfer.metadata.transfer_id == 14);
647691
REQUIRE(transfer.payload_size == 16);
648692
REQUIRE(0 == std::memcmp(transfer.payload, "\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07\x09\x09", 16));
649693
REQUIRE(ins.getAllocator().getNumAllocatedFragments() == 1);
@@ -652,18 +696,18 @@ TEST_CASE("rxSessionUpdate")
652696

653697
// Update another session using same frame.
654698
REQUIRE(1 == updateAnotherSession(1, 1'000'000, 16));
655-
REQUIRE(rxs_1.transfer_timestamp_usec == 20'000'100); // First frame.
699+
REQUIRE(rxs_1.transfer_timestamp_usec == 20'000'300); // The timestamp is the same as the first frame.
656700
REQUIRE(rxs_1.payload_size == 0);
657701
REQUIRE(rxs_1.payload == nullptr);
658702
REQUIRE(rxs_1.calculated_crc == 0xFFFFFFFFU);
659-
REQUIRE(rxs_1.transfer_id == 14U);
703+
REQUIRE(rxs_1.transfer_id == 15U);
660704
REQUIRE(rxs_1.redundant_transport_index == 1);
661-
REQUIRE(transfer.timestamp_usec == 20'000'100);
705+
REQUIRE(transfer.timestamp_usec == 20'000'300);
662706
REQUIRE(transfer.metadata.priority == UdpardPrioritySlow);
663707
REQUIRE(transfer.metadata.transfer_kind == UdpardTransferKindMessage);
664708
REQUIRE(transfer.metadata.port_id == 2'222);
665709
REQUIRE(transfer.metadata.remote_node_id == 55);
666-
REQUIRE(transfer.metadata.transfer_id == 13);
710+
REQUIRE(transfer.metadata.transfer_id == 14);
667711
REQUIRE(transfer.payload_size == 16);
668712
REQUIRE(0 == std::memcmp(transfer.payload, "\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07\x09\x09", 16));
669713
REQUIRE(ins_1.getAllocator().getNumAllocatedFragments() == 1);
@@ -678,7 +722,7 @@ TEST_CASE("rxSessionUpdate")
678722
frame.payload_size = 7;
679723
frame.payload = reinterpret_cast<const uint8_t*>("\x0A\x0A\x0A\x0A\x0A\x0A\x0A");
680724
REQUIRE(0 == update(2, 1'000'000, 16));
681-
REQUIRE(rxs.transfer_timestamp_usec == 20'000'100); // No change.
725+
REQUIRE(rxs.transfer_timestamp_usec == 20'000'300); // No change.
682726
REQUIRE(rxs.payload_size == 0);
683727
REQUIRE(rxs.payload == nullptr);
684728
REQUIRE(rxs.calculated_crc == 0xFFFFFFFFU);
@@ -688,14 +732,14 @@ TEST_CASE("rxSessionUpdate")
688732
REQUIRE(ins.getAllocator().getTotalAllocatedAmount() == 0);
689733

690734
// OOM -- reset on error.
691-
frame.timestamp_usec = 20'000'200;
735+
frame.timestamp_usec = 20'000'400;
692736
frame.transfer_id = 30;
693737
frame.start_of_transfer = true;
694738
frame.end_of_transfer = true;
695739
frame.payload_size = 8;
696740
frame.payload = reinterpret_cast<const uint8_t*>("\x0E\x0E\x0E\x0E\x0E\x0E\x0E\xF7");
697741
REQUIRE((-UDPARD_ERROR_OUT_OF_MEMORY) == update(2, 1'000'000, 17)); // Exceeds the heap quota.
698-
REQUIRE(rxs.transfer_timestamp_usec == 20'000'200);
742+
REQUIRE(rxs.transfer_timestamp_usec == 20'000'400);
699743
REQUIRE(rxs.payload_size == 0);
700744
REQUIRE(rxs.payload == nullptr);
701745
REQUIRE(rxs.calculated_crc == 0xFFFFFFFFU);

‎tests/test_public_roundtrip.cpp‎

Lines changed: 9 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -170,19 +170,15 @@ TEST_CASE("RoundtripSimple")
170170

171171
UdpardRxTransfer transfer{};
172172
UdpardRxSubscription* subscription = nullptr;
173-
std::int8_t result =
174-
ins_rx.rxAccept(ti->tx_deadline_usec, ti->frame, 0, ti->specifier, transfer, &subscription);
175-
REQUIRE(((-UDPARD_ERROR_OUT_OF_ORDER == ins_rx.rxAccept(ti->tx_deadline_usec,
176-
ti->frame,
177-
1,
178-
ti->specifier,
179-
transfer,
180-
&subscription)) || (0 == ins_rx.rxAccept(ti->tx_deadline_usec,
181-
ti->frame,
182-
1,
183-
ti->specifier,
184-
transfer,
185-
&subscription)))); // Redundant interface will never be used here.
173+
174+
std::int8_t result = ins_rx.rxAccept(
175+
ti->tx_deadline_usec,
176+
ti->frame,
177+
0,
178+
ti->specifier,
179+
transfer,
180+
&subscription);
181+
186182
if (result == 1)
187183
{
188184
Pending reference{}; // Fetch the reference transfer from the list of pending.

0 commit comments

Comments
 (0)