Skip to content

Commit 6307e1e

Browse files
pavel-kirienkopavel-kirienko
committed
udpard_user_context_t
1 parent 097d777 commit 6307e1e

File tree

10 files changed

+253
-124
lines changed

10 files changed

+253
-124
lines changed

‎libudpard/udpard.c‎

Lines changed: 40 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -608,15 +608,15 @@ typedef struct tx_transfer_t
608608
/// they contain the values encoded in the P2P header. This is needed to find pending acks (to minimize duplicates),
609609
/// and to report the correct values via the feedback callback for P2P transfers.
610610
/// By default, upon construction, the remote_* fields equal the local ones, which is valid for ordinary messages.
611-
uint64_t topic_hash;
612-
uint64_t transfer_id;
613-
uint64_t remote_topic_hash;
614-
uint64_t remote_transfer_id;
615-
udpard_us_t deadline;
616-
bool reliable;
617-
udpard_prio_t priority;
618-
udpard_udpip_ep_t destination[UDPARD_IFACE_COUNT_MAX];
619-
void* user_transfer_reference;
611+
uint64_t topic_hash;
612+
uint64_t transfer_id;
613+
uint64_t remote_topic_hash;
614+
uint64_t remote_transfer_id;
615+
udpard_us_t deadline;
616+
bool reliable;
617+
udpard_prio_t priority;
618+
udpard_udpip_ep_t destination[UDPARD_IFACE_COUNT_MAX];
619+
udpard_user_context_t user;
620620

621621
void (*feedback)(udpard_tx_t*, udpard_tx_feedback_t);
622622
} tx_transfer_t;
@@ -649,10 +649,12 @@ static void tx_transfer_free_payload(tx_transfer_t* const tr)
649649
static void tx_transfer_retire(udpard_tx_t* const tx, tx_transfer_t* const tr, const bool success)
650650
{
651651
// Construct the feedback object first before the transfer is destroyed.
652-
const udpard_tx_feedback_t fb = { .topic_hash = tr->remote_topic_hash,
653-
.transfer_id = tr->remote_transfer_id,
654-
.user_transfer_reference = tr->user_transfer_reference,
655-
.success = success };
652+
const udpard_tx_feedback_t fb = {
653+
.topic_hash = tr->remote_topic_hash,
654+
.transfer_id = tr->remote_transfer_id,
655+
.user = tr->user,
656+
.success = success,
657+
};
656658
UDPARD_ASSERT(tr->reliable == (tr->feedback != NULL));
657659
// save the feedback pointer
658660
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t) = tr->feedback;
@@ -876,8 +878,8 @@ static uint32_t tx_push(udpard_tx_t* const tx,
876878
const udpard_udpip_ep_t endpoint[UDPARD_IFACE_COUNT_MAX],
877879
const udpard_bytes_scattered_t payload,
878880
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t),
879-
void* const user_transfer_reference,
880-
tx_transfer_t** const out_transfer)
881+
const udpard_user_context_t user,
882+
tx_transfer_t** const out_transfer)
881883
{
882884
UDPARD_ASSERT(now <= deadline);
883885
UDPARD_ASSERT(tx != NULL);
@@ -900,17 +902,17 @@ static uint32_t tx_push(udpard_tx_t* const tx,
900902
return 0;
901903
}
902904
mem_zero(sizeof(*tr), tr);
903-
tr->epoch = 0;
904-
tr->staged_until = now;
905-
tr->topic_hash = meta.topic_hash;
906-
tr->transfer_id = meta.transfer_id;
907-
tr->remote_topic_hash = meta.topic_hash;
908-
tr->remote_transfer_id = meta.transfer_id;
909-
tr->deadline = deadline;
910-
tr->reliable = meta.flag_ack;
911-
tr->priority = meta.priority;
912-
tr->user_transfer_reference = user_transfer_reference;
913-
tr->feedback = feedback;
905+
tr->epoch = 0;
906+
tr->staged_until = now;
907+
tr->topic_hash = meta.topic_hash;
908+
tr->transfer_id = meta.transfer_id;
909+
tr->remote_topic_hash = meta.topic_hash;
910+
tr->remote_transfer_id = meta.transfer_id;
911+
tr->deadline = deadline;
912+
tr->reliable = meta.flag_ack;
913+
tr->priority = meta.priority;
914+
tr->user = user;
915+
tr->feedback = feedback;
914916
for (size_t i = 0; i < UDPARD_IFACE_COUNT_MAX; i++) {
915917
tr->destination[i] = endpoint[i];
916918
tr->head[i] = tr->cursor[i] = NULL;
@@ -1057,7 +1059,7 @@ static void tx_send_ack(udpard_rx_t* const rx,
10571059
remote.endpoints,
10581060
(udpard_bytes_scattered_t){ .bytes = payload, .next = NULL },
10591061
NULL,
1060-
NULL,
1062+
UDPARD_USER_CONTEXT_NULL,
10611063
&tr);
10621064
UDPARD_ASSERT(count <= 1);
10631065
if (count == 1) { // ack is always a single-frame transfer, so we get either 0 or 1
@@ -1119,7 +1121,7 @@ uint32_t udpard_tx_push(udpard_tx_t* const self,
11191121
const uint64_t transfer_id,
11201122
const udpard_bytes_scattered_t payload,
11211123
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t),
1122-
void* const user_transfer_reference)
1124+
const udpard_user_context_t user)
11231125
{
11241126
uint32_t out = 0;
11251127
const bool ok = (self != NULL) && (deadline >= now) && (now >= 0) && (self->local_uid != 0) &&
@@ -1136,7 +1138,7 @@ uint32_t udpard_tx_push(udpard_tx_t* const self,
11361138
.sender_uid = self->local_uid,
11371139
.topic_hash = topic_hash,
11381140
};
1139-
out = tx_push(self, now, deadline, meta, remote_ep, payload, feedback, user_transfer_reference, NULL);
1141+
out = tx_push(self, now, deadline, meta, remote_ep, payload, feedback, user, NULL);
11401142
}
11411143
return out;
11421144
}
@@ -1150,7 +1152,7 @@ uint32_t udpard_tx_push_p2p(udpard_tx_t* const self,
11501152
const udpard_remote_t remote,
11511153
const udpard_bytes_scattered_t payload,
11521154
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t),
1153-
void* const user_transfer_reference)
1155+
const udpard_user_context_t user)
11541156
{
11551157
uint32_t out = 0;
11561158
const bool ok = (self != NULL) && (deadline >= now) && (now >= 0) && (self->local_uid != 0) &&
@@ -1179,8 +1181,7 @@ uint32_t udpard_tx_push_p2p(udpard_tx_t* const self,
11791181
.topic_hash = remote.uid,
11801182
};
11811183
tx_transfer_t* tr = NULL;
1182-
out =
1183-
tx_push(self, now, deadline, meta, remote.endpoints, full_payload, feedback, user_transfer_reference, &tr);
1184+
out = tx_push(self, now, deadline, meta, remote.endpoints, full_payload, feedback, user, &tr);
11841185
if (out > 0) {
11851186
UDPARD_ASSERT(tr != NULL);
11861187
tr->remote_topic_hash = request_topic_hash;
@@ -1252,13 +1253,13 @@ static void tx_eject_pending_frames(udpard_tx_t* const self, const udpard_us_t n
12521253
const bool last_attempt = !cavl2_is_inserted(self->index_staged, &tr->index_staged);
12531254
const bool last_frame = frame_next == NULL; // if not last attempt we will have to rewind to head.
12541255
const udpard_tx_ejection_t ejection = {
1255-
.now = now,
1256-
.deadline = tr->deadline,
1257-
.iface_index = ifindex,
1258-
.dscp = self->dscp_value_per_priority[tr->priority],
1259-
.destination = tr->destination[ifindex],
1260-
.datagram = tx_frame_view(frame),
1261-
.user_transfer_reference = tr->user_transfer_reference,
1256+
.now = now,
1257+
.deadline = tr->deadline,
1258+
.iface_index = ifindex,
1259+
.dscp = self->dscp_value_per_priority[tr->priority],
1260+
.destination = tr->destination[ifindex],
1261+
.datagram = tx_frame_view(frame),
1262+
.user = tr->user,
12621263
};
12631264
if (!self->vtable->eject(self, ejection)) { // The easy case -- no progress was made at this time;
12641265
break; // don't change anything, just try again later as-is

‎libudpard/udpard.h‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,20 @@ typedef struct udpard_bytes_mut_t
158158
void* data;
159159
} udpard_bytes_mut_t;
160160

161+
/// The library carries the user-provided context from inputs to outputs without interpreting it,
162+
/// allowing the application to associate its own data with various entities inside the library.
163+
typedef struct udpard_user_context_t
164+
{
165+
void (*fun)(void);
166+
void* obj;
167+
} udpard_user_context_t;
168+
#if defined(__cplusplus)
169+
#define UDPARD_USER_CONTEXT_NULL \
170+
udpard_user_context_t {}
171+
#else
172+
#define UDPARD_USER_CONTEXT_NULL ((udpard_user_context_t){ .fun = NULL, .obj = NULL })
173+
#endif
174+
161175
/// Zeros if invalid/unset/unavailable.
162176
typedef struct udpard_udpip_ep_t
163177
{
@@ -335,7 +349,8 @@ typedef struct udpard_tx_feedback_t
335349
{
336350
uint64_t topic_hash;
337351
uint64_t transfer_id;
338-
void* user_transfer_reference; ///< This is the same pointer that was passed to udpard_tx_push().
352+
353+
udpard_user_context_t user; ///< Same value that was passed to udpard_tx_push().
339354

340355
bool success; ///< False if no ack was received from the remote end before deadline expiration or forced eviction.
341356
} udpard_tx_feedback_t;
@@ -362,8 +377,8 @@ typedef struct udpard_tx_ejection_t
362377
/// udpard_tx_refcount_dec() must be invoked to release the reference.
363378
udpard_bytes_t datagram;
364379

365-
/// This is the same pointer that was passed to udpard_tx_push().
366-
void* user_transfer_reference;
380+
/// This is the same value that was passed to udpard_tx_push().
381+
udpard_user_context_t user;
367382
} udpard_tx_ejection_t;
368383

369384
/// Virtual function table for the TX pipeline, to be provided by the application.
@@ -466,8 +481,7 @@ bool udpard_tx_new(udpard_tx_t* const self,
466481
/// such that it is likely to be distinct per application startup (embedded systems can use noinit memory sections,
467482
/// hash uninitialized SRAM, use timers or ADC noise, etc).
468483
///
469-
/// The user_transfer_reference is an opaque pointer that will be stored for each enqueued item of this transfer.
470-
/// The library itself does not use or check this value in any way, so it can be NULL if not needed.
484+
/// The user context value is carried through to the callbacks; use UDPARD_USER_CONTEXT_NULL if not needed.
471485
///
472486
/// The function returns the number of datagrams enqueued, which is always a positive number, on success.
473487
/// In case of failure, the function returns zero. Runtime failures increment the corresponding error counters,
@@ -498,7 +512,7 @@ uint32_t udpard_tx_push(udpard_tx_t* const self,
498512
const uint64_t transfer_id,
499513
const udpard_bytes_scattered_t payload,
500514
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t), // NULL if best-effort.
501-
void* const user_transfer_reference);
515+
const udpard_user_context_t user);
502516

503517
/// This is a specialization of the general push function for P2P transfers.
504518
/// It is used to send P2P responses to messages received from topics; the request_* values shall be taken from
@@ -516,7 +530,7 @@ uint32_t udpard_tx_push_p2p(udpard_tx_t* const self,
516530
const udpard_remote_t remote, // Endpoints may be invalid for some ifaces.
517531
const udpard_bytes_scattered_t payload,
518532
void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t), // NULL if best-effort.
519-
void* const user_transfer_reference);
533+
const udpard_user_context_t user);
520534

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

‎tests/src/helpers.h‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,13 @@ static inline udpard_bytes_scattered_t make_scattered(const void* const data, co
6464
return out;
6565
}
6666

67+
// Wraps an application pointer for user context plumbing.
68+
static inline udpard_user_context_t make_user_context(void* const obj)
69+
{
70+
udpard_user_context_t out = { .fun = NULL, .obj = obj };
71+
return out;
72+
}
73+
6774
/// The instrumented allocator tracks memory consumption, checks for heap corruption, and can be configured to fail
6875
/// allocations above a certain threshold.
6976
#define INSTRUMENTED_ALLOCATOR_CANARY_SIZE 1024U

‎tests/src/test_e2e_api.cpp‎

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@ constexpr udpard_tx_vtable_t tx_vtable{ .eject = &capture_tx_frame };
7272
// Feedback callback records completion.
7373
void record_feedback(udpard_tx_t*, const udpard_tx_feedback_t fb)
7474
{
75-
auto* st = static_cast<FeedbackState*>(fb.user_transfer_reference);
75+
auto* st = static_cast<FeedbackState*>(fb.user.obj);
7676
if (st != nullptr) {
7777
st->count++;
7878
st->success = fb.success;
@@ -235,7 +235,7 @@ void test_reliable_delivery_under_losses()
235235
1U,
236236
payload_view,
237237
&record_feedback,
238-
&fb));
238+
make_user_context(&fb)));
239239

240240
// Send until acked; drop first data frame and first ack.
241241
bool first_round = true;
@@ -341,10 +341,17 @@ void test_reliable_stats_and_failures()
341341
FeedbackState fb_fail{};
342342
const udpard_udpip_ep_t exp_dest[UDPARD_IFACE_COUNT_MAX] = { udpard_make_subject_endpoint(99U), {}, {} };
343343
const udpard_bytes_scattered_t exp_payload = make_scattered("ping", 4);
344-
TEST_ASSERT_GREATER_THAN_UINT32(
345-
0U,
346-
udpard_tx_push(
347-
&exp_tx, 0, 10, udpard_prio_fast, 0xABCULL, exp_dest, 5U, exp_payload, &record_feedback, &fb_fail));
344+
TEST_ASSERT_GREATER_THAN_UINT32(0U,
345+
udpard_tx_push(&exp_tx,
346+
0,
347+
10,
348+
udpard_prio_fast,
349+
0xABCULL,
350+
exp_dest,
351+
5U,
352+
exp_payload,
353+
&record_feedback,
354+
make_user_context(&fb_fail)));
348355
udpard_tx_poll(&exp_tx, 0, UDPARD_IFACE_MASK_ALL);
349356
for (const auto& f : exp_frames) {
350357
drop_frame(f);
@@ -395,10 +402,17 @@ void test_reliable_stats_and_failures()
395402
const udpard_udpip_ep_t src_dest[UDPARD_IFACE_COUNT_MAX] = { udpard_make_subject_endpoint(12U), {}, {} };
396403
const udpard_bytes_scattered_t src_payload = make_scattered(ctx.expected.data(), ctx.expected.size());
397404
FeedbackState fb_ignore{};
398-
TEST_ASSERT_GREATER_THAN_UINT32(
399-
0U,
400-
udpard_tx_push(
401-
&src_tx, 0, 1000, udpard_prio_fast, port.topic_hash, src_dest, 7U, src_payload, &record_feedback, &fb_ignore));
405+
TEST_ASSERT_GREATER_THAN_UINT32(0U,
406+
udpard_tx_push(&src_tx,
407+
0,
408+
1000,
409+
udpard_prio_fast,
410+
port.topic_hash,
411+
src_dest,
412+
7U,
413+
src_payload,
414+
&record_feedback,
415+
make_user_context(&fb_ignore)));
402416
udpard_tx_poll(&src_tx, 0, UDPARD_IFACE_MASK_ALL);
403417
const udpard_mem_deleter_t tx_payload_deleter{ .user = nullptr, .free = &tx_refcount_free };
404418
for (const auto& f : src_frames) {

0 commit comments

Comments
 (0)