Skip to content

Commit 6ae1709

Browse files
pavel-kirienkopavel-kirienko
committed
fix AVL ordering
1 parent 7eb603a commit 6ae1709

3 files changed

Lines changed: 118 additions & 42 deletions

File tree

‎libudpard/udpard.c‎

Lines changed: 74 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -595,8 +595,8 @@ typedef struct
595595
/// The transfer index contains ALL transfers, used for lookup by (topic_hash, transfer_id).
596596
typedef struct tx_transfer_t
597597
{
598-
udpard_tree_t index_staged; ///< Soonest to be ready on the left. Key: staged_until
599-
udpard_tree_t index_deadline; ///< Soonest to expire on the left. Key: deadline
598+
udpard_tree_t index_staged; ///< Soonest to be ready on the left. Key: staged_until + transfer identity
599+
udpard_tree_t index_deadline; ///< Soonest to expire on the left. Key: deadline + transfer identity
600600
udpard_tree_t index_transfer; ///< Specific transfer lookup for ack management. Key: tx_transfer_key_t
601601
udpard_listed_t queue[UDPARD_IFACE_COUNT_MAX]; ///< Listed when ready for transmission.
602602
udpard_listed_t agewise; ///< Listed when created; oldest at the tail.
@@ -734,13 +734,40 @@ static bool tx_ensure_queue_space(udpard_tx_t* const tx, const size_t total_fram
734734
return total_frames_needed <= (tx->enqueued_frames_limit - tx->enqueued_frames_count);
735735
}
736736

737+
// Key for time-ordered TX indices with stable tiebreaking.
738+
typedef struct
739+
{
740+
udpard_us_t time;
741+
uint64_t topic_hash;
742+
uint64_t transfer_id;
743+
} tx_time_key_t;
744+
745+
// Compare staged transfers by time then by transfer identity.
737746
static int32_t tx_cavl_compare_staged(const void* const user, const udpard_tree_t* const node)
738747
{
739-
return ((*(const udpard_us_t*)user) >= CAVL2_TO_OWNER(node, tx_transfer_t, index_staged)->staged_until) ? +1 : -1;
748+
const tx_time_key_t* const key = (const tx_time_key_t*)user;
749+
const tx_transfer_t* const tr = CAVL2_TO_OWNER(node, tx_transfer_t, index_staged); // clang-format off
750+
if (key->time < tr->staged_until) { return -1; }
751+
if (key->time > tr->staged_until) { return +1; }
752+
if (key->topic_hash < tr->topic_hash) { return -1; }
753+
if (key->topic_hash > tr->topic_hash) { return +1; }
754+
if (key->transfer_id < tr->transfer_id) { return -1; }
755+
if (key->transfer_id > tr->transfer_id) { return +1; }
756+
return 0; // clang-format on
740757
}
758+
759+
// Compare deadlines by time then by transfer identity.
741760
static int32_t tx_cavl_compare_deadline(const void* const user, const udpard_tree_t* const node)
742761
{
743-
return ((*(const udpard_us_t*)user) >= CAVL2_TO_OWNER(node, tx_transfer_t, index_deadline)->deadline) ? +1 : -1;
762+
const tx_time_key_t* const key = (const tx_time_key_t*)user;
763+
const tx_transfer_t* const tr = CAVL2_TO_OWNER(node, tx_transfer_t, index_deadline); // clang-format off
764+
if (key->time < tr->deadline) { return -1; }
765+
if (key->time > tr->deadline) { return +1; }
766+
if (key->topic_hash < tr->topic_hash) { return -1; }
767+
if (key->topic_hash > tr->topic_hash) { return +1; }
768+
if (key->transfer_id < tr->transfer_id) { return -1; }
769+
if (key->transfer_id > tr->transfer_id) { return +1; }
770+
return 0; // clang-format on
744771
}
745772
static int32_t tx_cavl_compare_transfer(const void* const user, const udpard_tree_t* const node)
746773
{
@@ -855,11 +882,18 @@ static void tx_stage_if(udpard_tx_t* const tx, tx_transfer_t* const tr)
855882
const udpard_us_t timeout = tx_ack_timeout(tx->ack_baseline_timeout, tr->priority, epoch);
856883
tr->staged_until += timeout;
857884
if ((tr->deadline - timeout) >= tr->staged_until) {
858-
(void)cavl2_find_or_insert(&tx->index_staged, //
859-
&tr->staged_until,
860-
tx_cavl_compare_staged,
861-
&tr->index_staged,
862-
cavl2_trivial_factory);
885+
// Insert into staged index with deterministic tie-breaking.
886+
const tx_time_key_t key = { .time = tr->staged_until,
887+
.topic_hash = tr->topic_hash,
888+
.transfer_id = tr->transfer_id };
889+
// Ensure we didn't collide with another entry that should be unique.
890+
const udpard_tree_t* const tree_staged = cavl2_find_or_insert(&tx->index_staged, //
891+
&key,
892+
tx_cavl_compare_staged,
893+
&tr->index_staged,
894+
cavl2_trivial_factory);
895+
UDPARD_ASSERT(tree_staged == &tr->index_staged);
896+
(void)tree_staged;
863897
}
864898
}
865899

@@ -1049,12 +1083,19 @@ static bool tx_push(udpard_tx_t* const tx,
10491083
tx_stage_if(tx, tr);
10501084
}
10511085
// Add to the deadline index for expiration management.
1052-
(void)cavl2_find_or_insert(
1053-
&tx->index_deadline, &tr->deadline, tx_cavl_compare_deadline, &tr->index_deadline, cavl2_trivial_factory);
1086+
// Insert into deadline index with deterministic tie-breaking.
1087+
const tx_time_key_t deadline_key = { .time = tr->deadline,
1088+
.topic_hash = tr->topic_hash,
1089+
.transfer_id = tr->transfer_id };
1090+
// Ensure we didn't collide with another entry that should be unique.
1091+
const udpard_tree_t* const tree_deadline = cavl2_find_or_insert(
1092+
&tx->index_deadline, &deadline_key, tx_cavl_compare_deadline, &tr->index_deadline, cavl2_trivial_factory);
1093+
UDPARD_ASSERT(tree_deadline == &tr->index_deadline);
1094+
(void)tree_deadline;
10541095
// Add to the transfer index for incoming ack management.
1055-
const tx_transfer_key_t key = { .topic_hash = tr->topic_hash, .transfer_id = tr->transfer_id };
1096+
const tx_transfer_key_t transfer_key = { .topic_hash = tr->topic_hash, .transfer_id = tr->transfer_id };
10561097
const udpard_tree_t* const tree_transfer = cavl2_find_or_insert(
1057-
&tx->index_transfer, &key, tx_cavl_compare_transfer, &tr->index_transfer, cavl2_trivial_factory);
1098+
&tx->index_transfer, &transfer_key, tx_cavl_compare_transfer, &tr->index_transfer, cavl2_trivial_factory);
10581099
UDPARD_ASSERT(tree_transfer == &tr->index_transfer); // ensure no duplicates; checked at the API level
10591100
(void)tree_transfer;
10601101
// Add to the agewise list for sacrifice management on queue exhaustion.
@@ -1907,11 +1948,23 @@ static int32_t cavl_compare_rx_session_by_remote_uid(const void* const user, con
19071948
return 0; // clang-format on
19081949
}
19091950

1951+
// Key for reordering deadline ordering with stable tiebreaking.
1952+
typedef struct
1953+
{
1954+
udpard_us_t deadline;
1955+
uint64_t remote_uid;
1956+
} rx_reordering_key_t;
1957+
1958+
// Compare sessions by reordering deadline then by remote UID.
19101959
static int32_t cavl_compare_rx_session_by_reordering_deadline(const void* const user, const udpard_tree_t* const node)
19111960
{
1912-
const udpard_us_t dl_a = *(const udpard_us_t*)user;
1913-
const udpard_us_t dl_b = CAVL2_TO_OWNER(node, rx_session_t, index_reordering_window)->reordering_window_deadline;
1914-
return (dl_a >= dl_b) ? +1 : -1;
1961+
const rx_reordering_key_t* const key = (const rx_reordering_key_t*)user;
1962+
const rx_session_t* const ses = CAVL2_TO_OWNER(node, rx_session_t, index_reordering_window); // clang-format off
1963+
if (key->deadline < ses->reordering_window_deadline) { return -1; }
1964+
if (key->deadline > ses->reordering_window_deadline) { return +1; }
1965+
if (key->remote_uid < ses->remote.uid) { return -1; }
1966+
if (key->remote_uid > ses->remote.uid) { return +1; }
1967+
return 0; // clang-format on
19151968
}
19161969

19171970
typedef struct
@@ -2028,8 +2081,11 @@ static void rx_session_ordered_scan_slots(rx_session_t* const self,
20282081
// closure deadline, but we ignore them because the nearest transfer overrides the more distant ones.
20292082
if (slot != NULL) {
20302083
self->reordering_window_deadline = slot->ts_min + self->port->reordering_window;
2031-
const udpard_tree_t* res = cavl2_find_or_insert(&rx->index_session_by_reordering, //-------------
2032-
&self->reordering_window_deadline,
2084+
// Insert into reordering index with deterministic tie-breaking.
2085+
const rx_reordering_key_t key = { .deadline = self->reordering_window_deadline,
2086+
.remote_uid = self->remote.uid };
2087+
const udpard_tree_t* res = cavl2_find_or_insert(&rx->index_session_by_reordering, //----------------
2088+
&key,
20332089
&cavl_compare_rx_session_by_reordering_deadline,
20342090
&self->index_reordering_window,
20352091
&cavl2_trivial_factory);

‎tests/src/test_intrusive_rx.c‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2916,7 +2916,8 @@ static void test_rx_additional_coverage(void)
29162916
ses->slots[0].state = rx_slot_done;
29172917
ses->slots[0].transfer_id = 5;
29182918
TEST_ASSERT_TRUE(rx_session_is_transfer_interned(ses, 5));
2919-
udpard_us_t dl_key = 5;
2919+
// Comparator smoke-test with stable key.
2920+
const rx_reordering_key_t dl_key = { .deadline = 5, .remote_uid = ses->remote.uid };
29202921
(void)cavl_compare_rx_session_by_reordering_deadline(&dl_key, &ses->index_reordering_window);
29212922
udpard_list_t anim_list = { 0 };
29222923
udpard_tree_t* by_reorder = NULL;
@@ -2925,12 +2926,15 @@ static void test_rx_additional_coverage(void)
29252926
cavl_compare_rx_session_by_remote_uid,
29262927
&ses->index_remote_uid,
29272928
cavl2_trivial_factory);
2928-
ses->reordering_window_deadline = 3;
2929-
cavl2_find_or_insert(&by_reorder,
2930-
&ses->reordering_window_deadline,
2931-
cavl_compare_rx_session_by_reordering_deadline,
2932-
&ses->index_reordering_window,
2933-
cavl2_trivial_factory);
2929+
ses->reordering_window_deadline = 3;
2930+
const rx_reordering_key_t reorder_key = { .deadline = ses->reordering_window_deadline,
2931+
.remote_uid = ses->remote.uid };
2932+
const udpard_tree_t* const tree_reorder = cavl2_find_or_insert(&by_reorder,
2933+
&reorder_key,
2934+
cavl_compare_rx_session_by_reordering_deadline,
2935+
&ses->index_reordering_window,
2936+
cavl2_trivial_factory);
2937+
TEST_ASSERT_EQUAL_PTR(&ses->index_reordering_window, tree_reorder);
29342938
enlist_head(&anim_list, &ses->list_by_animation);
29352939
rx_session_free(ses, &anim_list, &by_reorder);
29362940

‎tests/src/test_intrusive_tx.c‎

Lines changed: 33 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -208,10 +208,17 @@ static void test_tx_validation_and_free(void)
208208
tr->remote_topic_hash = 99;
209209
tr->remote_transfer_id = 100;
210210
tx_transfer_key_t key = { .topic_hash = 5, .transfer_id = 7 };
211+
// Insert with stable ordering keys.
212+
const tx_time_key_t staged_key = { .time = tr->staged_until,
213+
.topic_hash = tr->topic_hash,
214+
.transfer_id = tr->transfer_id };
215+
const tx_time_key_t deadline_key = { .time = tr->deadline,
216+
.topic_hash = tr->topic_hash,
217+
.transfer_id = tr->transfer_id };
211218
(void)cavl2_find_or_insert(
212-
&tx.index_staged, &tr->staged_until, tx_cavl_compare_staged, &tr->index_staged, cavl2_trivial_factory);
219+
&tx.index_staged, &staged_key, tx_cavl_compare_staged, &tr->index_staged, cavl2_trivial_factory);
213220
(void)cavl2_find_or_insert(
214-
&tx.index_deadline, &tr->deadline, tx_cavl_compare_deadline, &tr->index_deadline, cavl2_trivial_factory);
221+
&tx.index_deadline, &deadline_key, tx_cavl_compare_deadline, &tr->index_deadline, cavl2_trivial_factory);
215222
(void)cavl2_find_or_insert(
216223
&tx.index_transfer, &key, tx_cavl_compare_transfer, &tr->index_transfer, cavl2_trivial_factory);
217224
(void)cavl2_find_or_insert(
@@ -236,14 +243,14 @@ static void test_tx_comparators_and_feedback(void)
236243
tr.remote_transfer_id = 4;
237244

238245
// Staged/deadline comparisons both ways.
239-
udpard_us_t us = 6;
240-
TEST_ASSERT_EQUAL(1, tx_cavl_compare_staged(&us, &tr.index_staged));
241-
us = 4;
242-
TEST_ASSERT_EQUAL(-1, tx_cavl_compare_staged(&us, &tr.index_staged));
243-
us = 8;
244-
TEST_ASSERT_EQUAL(1, tx_cavl_compare_deadline(&us, &tr.index_deadline));
245-
us = 6;
246-
TEST_ASSERT_EQUAL(-1, tx_cavl_compare_deadline(&us, &tr.index_deadline));
246+
tx_time_key_t tkey = { .time = 6, .topic_hash = tr.topic_hash, .transfer_id = tr.transfer_id };
247+
TEST_ASSERT_EQUAL(1, tx_cavl_compare_staged(&tkey, &tr.index_staged));
248+
tkey.time = 4;
249+
TEST_ASSERT_EQUAL(-1, tx_cavl_compare_staged(&tkey, &tr.index_staged));
250+
tkey.time = 8;
251+
TEST_ASSERT_EQUAL(1, tx_cavl_compare_deadline(&tkey, &tr.index_deadline));
252+
tkey.time = 6;
253+
TEST_ASSERT_EQUAL(-1, tx_cavl_compare_deadline(&tkey, &tr.index_deadline));
247254

248255
// Transfer comparator covers all branches.
249256
tx_transfer_key_t key = { .topic_hash = 5, .transfer_id = 1 };
@@ -334,11 +341,12 @@ static void test_tx_spool_and_queue_errors(void)
334341
victim.deadline = 1;
335342
victim.topic_hash = 7;
336343
victim.transfer_id = 9;
337-
(void)cavl2_find_or_insert(&tx_sac.index_deadline,
338-
&victim.deadline,
339-
tx_cavl_compare_deadline,
340-
&victim.index_deadline,
341-
cavl2_trivial_factory);
344+
// Insert into deadline index with stable key.
345+
const tx_time_key_t deadline_key = { .time = victim.deadline,
346+
.topic_hash = victim.topic_hash,
347+
.transfer_id = victim.transfer_id };
348+
(void)cavl2_find_or_insert(
349+
&tx_sac.index_deadline, &deadline_key, tx_cavl_compare_deadline, &victim.index_deadline, cavl2_trivial_factory);
342350
(void)cavl2_find_or_insert(
343351
&tx_sac.index_transfer,
344352
&(tx_transfer_key_t){ .topic_hash = victim.topic_hash, .transfer_id = victim.transfer_id },
@@ -525,8 +533,12 @@ static void test_tx_ack_and_scheduler(void)
525533
exp->user = make_user_context(&fstate);
526534
exp->reliable = true;
527535
exp->feedback = record_feedback;
536+
// Insert into deadline index with stable key.
537+
const tx_time_key_t tx4_deadline_key = { .time = exp->deadline,
538+
.topic_hash = exp->topic_hash,
539+
.transfer_id = exp->transfer_id };
528540
(void)cavl2_find_or_insert(
529-
&tx4.index_deadline, &exp->deadline, tx_cavl_compare_deadline, &exp->index_deadline, cavl2_trivial_factory);
541+
&tx4.index_deadline, &tx4_deadline_key, tx_cavl_compare_deadline, &exp->index_deadline, cavl2_trivial_factory);
530542
(void)cavl2_find_or_insert(&tx4.index_transfer,
531543
&(tx_transfer_key_t){ .topic_hash = 55, .transfer_id = 66 },
532544
tx_cavl_compare_transfer,
@@ -554,8 +566,12 @@ static void test_tx_ack_and_scheduler(void)
554566
staged.p2p_destination[0] = make_ep(7);
555567
tx_frame_t dummy_frame = { 0 };
556568
staged.head[0] = staged.cursor[0] = &dummy_frame;
569+
// Insert into staged index with stable key.
570+
const tx_time_key_t tx5_staged_key = { .time = staged.staged_until,
571+
.topic_hash = staged.topic_hash,
572+
.transfer_id = staged.transfer_id };
557573
cavl2_find_or_insert(
558-
&tx5.index_staged, &staged.staged_until, tx_cavl_compare_staged, &staged.index_staged, cavl2_trivial_factory);
574+
&tx5.index_staged, &tx5_staged_key, tx_cavl_compare_staged, &staged.index_staged, cavl2_trivial_factory);
559575
tx5.ack_baseline_timeout = 1;
560576
tx_promote_staged_transfers(&tx5, 1);
561577
TEST_ASSERT_NOT_NULL(tx5.queue[0][staged.priority].head);

0 commit comments

Comments
 (0)