Skip to content

Commit 6b517d8

Browse files
pavel-kirienkopavel-kirienko
committed
add out transfer ID for P2P
1 parent 321d3af commit 6b517d8

6 files changed

Lines changed: 76 additions & 37 deletions

File tree

‎libudpard/udpard.c‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1198,7 +1198,8 @@ bool udpard_tx_push_p2p(udpard_tx_t* const self,
11981198
const udpard_remote_t remote,
11991199
const udpard_bytes_scattered_t payload,
12001200
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t),
1201-
const udpard_user_context_t user)
1201+
const udpard_user_context_t user,
1202+
uint64_t* const out_transfer_id)
12021203
{
12031204
const uint16_t iface_bitmap = valid_ep_bitmap(remote.endpoints);
12041205
bool ok = (self != NULL) && (deadline >= now) && (now >= 0) && (self->local_uid != 0) && (iface_bitmap != 0) &&
@@ -1240,6 +1241,10 @@ bool udpard_tx_push_p2p(udpard_tx_t* const self,
12401241
UDPARD_ASSERT(tr != NULL);
12411242
tr->remote_topic_hash = request_topic_hash;
12421243
tr->remote_transfer_id = request_transfer_id;
1244+
UDPARD_ASSERT(tr->transfer_id == meta.transfer_id);
1245+
if (out_transfer_id != NULL) {
1246+
*out_transfer_id = tr->transfer_id;
1247+
}
12431248
}
12441249
}
12451250
return ok;

‎libudpard/udpard.h‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -547,6 +547,7 @@ bool udpard_tx_push(udpard_tx_t* const self,
547547
/// It is used to send P2P responses to messages received from topics; the request_* values shall be taken from
548548
/// the message transfer that is being responded to. The topic_hash and the transfer_id fields of the feedback struct
549549
/// will be set to the request_topic_hash and request_transfer_id values, respectively.
550+
/// If out_transfer_id is not NULL, the assigned internal transfer-ID is stored there for use with udpard_tx_cancel_p2p.
550551
/// P2P transfers are a bit more complex because they carry some additional metadata that is automatically
551552
/// composed/parsed by the library transparently for the application.
552553
/// The size of the serialized payload will include UDPARD_P2P_HEADER_BYTES additional bytes for the P2P header.
@@ -559,7 +560,8 @@ bool udpard_tx_push_p2p(udpard_tx_t* const self,
559560
const udpard_remote_t remote, // Endpoints may be invalid for some ifaces.
560561
const udpard_bytes_scattered_t payload,
561562
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t), // NULL if best-effort.
562-
const udpard_user_context_t user);
563+
const udpard_user_context_t user,
564+
uint64_t* const out_transfer_id);
563565

564566
/// This should be invoked whenever the socket/NIC of this queue becomes ready to accept new datagrams for transmission.
565567
/// It is fine to also invoke it periodically unconditionally to drive the transmission process.

‎tests/src/test_e2e_edge.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -503,7 +503,8 @@ void test_udpard_tx_push_p2p()
503503
remote,
504504
payload,
505505
nullptr,
506-
UDPARD_USER_CONTEXT_NULL));
506+
UDPARD_USER_CONTEXT_NULL,
507+
nullptr));
507508
udpard_tx_poll(&tx, now, UDPARD_IFACE_BITMAP_ALL);
508509
TEST_ASSERT_FALSE(frames.empty());
509510

‎tests/src/test_e2e_responses.cpp‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -386,7 +386,8 @@ void test_topic_with_p2p_response()
386386
remote_a,
387387
response_scat,
388388
&record_feedback,
389-
make_user_context(&b_response_fb)));
389+
make_user_context(&b_response_fb),
390+
nullptr));
390391

391392
b_frames.clear();
392393
udpard_tx_poll(&b_tx, now, UDPARD_IFACE_BITMAP_ALL);
@@ -684,7 +685,8 @@ void test_topic_with_p2p_response_under_loss()
684685
remote_a,
685686
response_scat,
686687
&record_feedback,
687-
make_user_context(&b_response_fb)));
688+
make_user_context(&b_response_fb),
689+
nullptr));
688690
}
689691

690692
// --- Node B transmits (responses) ---

‎tests/src/test_intrusive_guards.c‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -159,8 +159,17 @@ static void test_tx_guards(void)
159159
&tx, 10, 5, iface_bitmap_1, udpard_prio_fast, 1U, 1U, empty_payload, NULL, UDPARD_USER_CONTEXT_NULL));
160160
TEST_ASSERT_FALSE(udpard_tx_push(
161161
NULL, 0, 0, iface_bitmap_1, udpard_prio_fast, 1U, 1U, empty_payload, NULL, UDPARD_USER_CONTEXT_NULL));
162-
TEST_ASSERT_FALSE(udpard_tx_push_p2p(
163-
NULL, 0, 0, udpard_prio_fast, 1U, 1U, (udpard_remote_t){ 0 }, empty_payload, NULL, UDPARD_USER_CONTEXT_NULL));
162+
TEST_ASSERT_FALSE(udpard_tx_push_p2p(NULL,
163+
0,
164+
0,
165+
udpard_prio_fast,
166+
1U,
167+
1U,
168+
(udpard_remote_t){ 0 },
169+
empty_payload,
170+
NULL,
171+
UDPARD_USER_CONTEXT_NULL,
172+
NULL));
164173

165174
// Poll and refcount no-ops on null data.
166175
udpard_tx_poll(NULL, 0, 0);

‎tests/src/test_intrusive_tx.c‎

Lines changed: 50 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -909,11 +909,9 @@ static void test_tx_cancel_p2p(void)
909909
TEST_ASSERT_TRUE(udpard_tx_new(&tx, 50U, 100U, 8U, mem, &vt));
910910
tx.user = &eject;
911911

912-
// Save the initial P2P transfer-ID counter.
913-
const uint64_t p2p_tid_initial = tx.p2p_transfer_id;
914-
915-
// Push a P2P transfer with feedback (reliable).
916-
const udpard_remote_t remote = { .uid = 999, .endpoints = { make_ep(10), make_ep(20) } };
912+
// Push a P2P transfer with feedback (reliable), using out_transfer_id.
913+
const udpard_remote_t remote = { .uid = 999, .endpoints = { make_ep(10), make_ep(20) } };
914+
uint64_t out_tid = 0;
917915
TEST_ASSERT_TRUE(udpard_tx_push_p2p(&tx,
918916
0,
919917
1000,
@@ -923,25 +921,28 @@ static void test_tx_cancel_p2p(void)
923921
remote,
924922
make_scattered(NULL, 0),
925923
record_feedback,
926-
make_user_context(&fstate)));
927-
TEST_ASSERT_EQUAL_UINT64(p2p_tid_initial + 1, tx.p2p_transfer_id);
924+
make_user_context(&fstate),
925+
&out_tid));
926+
// Verify out_transfer_id matches internal counter.
927+
TEST_ASSERT_EQUAL_UINT64(100, out_tid); // initial p2p_transfer_id was 100
928928
// P2P transfers are indexed by (remote.uid, internal_transfer_id).
929-
TEST_ASSERT_NOT_NULL(tx_transfer_find(&tx, remote.uid, p2p_tid_initial));
929+
TEST_ASSERT_NOT_NULL(tx_transfer_find(&tx, remote.uid, out_tid));
930930

931-
// Cancel using the correct destination_uid and internal transfer_id.
932-
TEST_ASSERT_TRUE(udpard_tx_cancel_p2p(&tx, remote.uid, p2p_tid_initial));
933-
TEST_ASSERT_NULL(tx_transfer_find(&tx, remote.uid, p2p_tid_initial));
931+
// Cancel using the returned transfer_id from out_transfer_id.
932+
TEST_ASSERT_TRUE(udpard_tx_cancel_p2p(&tx, remote.uid, out_tid));
933+
TEST_ASSERT_NULL(tx_transfer_find(&tx, remote.uid, out_tid));
934934
TEST_ASSERT_EQUAL_size_t(1, fstate.count);
935935
TEST_ASSERT_EQUAL_UINT32(0, fstate.last.acknowledgements);
936936
// Feedback returns request metadata, not internal P2P metadata.
937937
TEST_ASSERT_EQUAL_UINT64(0xABCD, fstate.last.topic_hash);
938938
TEST_ASSERT_EQUAL_UINT64(42, fstate.last.transfer_id);
939939

940940
// Cancelling again returns false (already cancelled).
941-
TEST_ASSERT_FALSE(udpard_tx_cancel_p2p(&tx, remote.uid, p2p_tid_initial));
941+
TEST_ASSERT_FALSE(udpard_tx_cancel_p2p(&tx, remote.uid, out_tid));
942942

943943
// Cancelling with wrong destination_uid returns false.
944-
fstate.count = 0;
944+
fstate.count = 0;
945+
uint64_t out_tid2 = 0;
945946
TEST_ASSERT_TRUE(udpard_tx_push_p2p(&tx,
946947
0,
947948
1000,
@@ -951,37 +952,56 @@ static void test_tx_cancel_p2p(void)
951952
remote,
952953
make_scattered(NULL, 0),
953954
record_feedback,
954-
make_user_context(&fstate)));
955-
const uint64_t p2p_tid_second = tx.p2p_transfer_id - 1;
956-
TEST_ASSERT_FALSE(udpard_tx_cancel_p2p(&tx, 888, p2p_tid_second)); // wrong destination
957-
TEST_ASSERT_NOT_NULL(tx_transfer_find(&tx, remote.uid, p2p_tid_second));
955+
make_user_context(&fstate),
956+
&out_tid2));
957+
TEST_ASSERT_FALSE(udpard_tx_cancel_p2p(&tx, 888, out_tid2)); // wrong destination
958+
TEST_ASSERT_NOT_NULL(tx_transfer_find(&tx, remote.uid, out_tid2));
958959
TEST_ASSERT_EQUAL_size_t(0, fstate.count); // feedback not invoked
959960

960961
// Cancelling with wrong transfer_id returns false.
961962
TEST_ASSERT_FALSE(udpard_tx_cancel_p2p(&tx, remote.uid, 12345)); // wrong transfer_id
962-
TEST_ASSERT_NOT_NULL(tx_transfer_find(&tx, remote.uid, p2p_tid_second));
963+
TEST_ASSERT_NOT_NULL(tx_transfer_find(&tx, remote.uid, out_tid2));
963964

964965
// Cancel with correct parameters works.
965-
TEST_ASSERT_TRUE(udpard_tx_cancel_p2p(&tx, remote.uid, p2p_tid_second));
966-
TEST_ASSERT_NULL(tx_transfer_find(&tx, remote.uid, p2p_tid_second));
966+
TEST_ASSERT_TRUE(udpard_tx_cancel_p2p(&tx, remote.uid, out_tid2));
967+
TEST_ASSERT_NULL(tx_transfer_find(&tx, remote.uid, out_tid2));
967968
TEST_ASSERT_EQUAL_size_t(1, fstate.count);
968969

969-
// Best-effort P2P transfer cancels quietly (no feedback).
970-
fstate.count = 0;
971-
TEST_ASSERT_TRUE(udpard_tx_push_p2p(
972-
&tx, 0, 1000, udpard_prio_nominal, 0x1234, 77, remote, make_scattered(NULL, 0), NULL, UDPARD_USER_CONTEXT_NULL));
973-
const uint64_t p2p_tid_third = tx.p2p_transfer_id - 1;
974-
TEST_ASSERT_TRUE(udpard_tx_cancel_p2p(&tx, remote.uid, p2p_tid_third));
970+
// Best-effort P2P transfer cancels quietly (no feedback); using NULL for out_transfer_id.
971+
fstate.count = 0;
972+
uint64_t out_tid3 = 0;
973+
TEST_ASSERT_TRUE(udpard_tx_push_p2p(&tx,
974+
0,
975+
1000,
976+
udpard_prio_nominal,
977+
0x1234,
978+
77,
979+
remote,
980+
make_scattered(NULL, 0),
981+
NULL,
982+
UDPARD_USER_CONTEXT_NULL,
983+
&out_tid3));
984+
TEST_ASSERT_TRUE(udpard_tx_cancel_p2p(&tx, remote.uid, out_tid3));
975985
TEST_ASSERT_EQUAL_size_t(0, fstate.count); // no feedback for best-effort
976986

977987
// Test cancel_all with P2P transfers (using destination_uid as topic_hash).
988+
// Pass NULL for out_transfer_id to verify optional behavior.
978989
TEST_ASSERT_TRUE(udpard_tx_push_p2p(
979-
&tx, 0, 1000, udpard_prio_fast, 0xA, 1, remote, make_scattered(NULL, 0), NULL, UDPARD_USER_CONTEXT_NULL));
990+
&tx, 0, 1000, udpard_prio_fast, 0xA, 1, remote, make_scattered(NULL, 0), NULL, UDPARD_USER_CONTEXT_NULL, NULL));
980991
TEST_ASSERT_TRUE(udpard_tx_push_p2p(
981-
&tx, 0, 1000, udpard_prio_fast, 0xB, 2, remote, make_scattered(NULL, 0), NULL, UDPARD_USER_CONTEXT_NULL));
992+
&tx, 0, 1000, udpard_prio_fast, 0xB, 2, remote, make_scattered(NULL, 0), NULL, UDPARD_USER_CONTEXT_NULL, NULL));
982993
const udpard_remote_t other_remote = { .uid = 777, .endpoints = { make_ep(30) } };
983-
TEST_ASSERT_TRUE(udpard_tx_push_p2p(
984-
&tx, 0, 1000, udpard_prio_fast, 0xC, 3, other_remote, make_scattered(NULL, 0), NULL, UDPARD_USER_CONTEXT_NULL));
994+
TEST_ASSERT_TRUE(udpard_tx_push_p2p(&tx,
995+
0,
996+
1000,
997+
udpard_prio_fast,
998+
0xC,
999+
3,
1000+
other_remote,
1001+
make_scattered(NULL, 0),
1002+
NULL,
1003+
UDPARD_USER_CONTEXT_NULL,
1004+
NULL));
9851005
TEST_ASSERT_EQUAL_size_t(2, udpard_tx_cancel_all(&tx, remote.uid));
9861006
TEST_ASSERT_EQUAL_size_t(1, udpard_tx_cancel_all(&tx, other_remote.uid));
9871007

0 commit comments

Comments
 (0)