udpard_tx_push_p2p

Author: pavel-kirienko

libudpard/udpard.c

@@ -890,9 +890,11 @@ static uint32_t tx_push(udpard_tx_t* const      tx,
     (void)cavl2_find_or_insert(
       &tx->index_deadline, &tr->deadline, tx_cavl_compare_deadline, &tr->index_deadline, cavl2_trivial_factory);
     // Add to the transfer index for incoming ack management.
-    const tx_transfer_key_t key = { .topic_hash = tr->topic_hash, .transfer_id = tr->transfer_id };
-    (void)cavl2_find_or_insert(
+    const tx_transfer_key_t    key           = { .topic_hash = tr->topic_hash, .transfer_id = tr->transfer_id };
+    const udpard_tree_t* const tree_transfer = cavl2_find_or_insert(
       &tx->index_transfer, &key, tx_cavl_compare_transfer, &tr->index_transfer, cavl2_trivial_factory);
+    UDPARD_ASSERT(tree_transfer == &tr->index_transfer); // ensure no duplicates; checked at the API level
+    (void)tree_transfer;
     // Add to the agewise list to allow instant sacrifice when needed; oldest at the tail.
     enlist_head(&tx->agewise, &tr->agewise);
 
@@ -1039,23 +1041,38 @@ uint32_t udpard_tx_push(udpard_tx_t* const      self,
         // Before attempting to enqueue a new transfer, we need to update the transmission scheduler.
         // It may release some items from the tx queue, and it may also promote some staged transfers to the queue.
         udpard_tx_poll(self, now, UDPARD_IFACE_MASK_ALL);
-        const meta_t meta = {
-            .priority              = priority,
-            .flag_ack              = feedback != NULL,
-            .transfer_payload_size = (uint32_t)payload.size,
-            .transfer_id           = transfer_id,
-            .sender_uid            = self->local_uid,
-            .topic_hash            = topic_hash,
-        };
-        out = tx_push(self, //
-                      now,
-                      deadline,
-                      meta,
-                      remote_ep,
-                      payload,
-                      feedback,
-                      user_transfer_reference,
-                      NULL);
+        const meta_t meta = { .priority              = priority,
+                              .flag_ack              = feedback != NULL,
+                              .transfer_payload_size = (uint32_t)payload.size,
+                              .transfer_id           = transfer_id,
+                              .sender_uid            = self->local_uid,
+                              .topic_hash            = topic_hash };
+        out = tx_push(self, now, deadline, meta, remote_ep, payload, feedback, user_transfer_reference, NULL);
+    }
+    return out;
+}
+
+uint32_t udpard_tx_push_p2p(udpard_tx_t* const    self,
+                            const udpard_us_t     now,
+                            const udpard_us_t     deadline,
+                            const udpard_prio_t   priority,
+                            const udpard_remote_t remote,
+                            const udpard_bytes_t  payload,
+                            void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t),
+                            void* const user_transfer_reference)
+{
+    uint32_t out = 0;
+    if (self != NULL) {
+        out = udpard_tx_push(self,
+                             now,
+                             deadline,
+                             priority,
+                             remote.uid,
+                             remote.endpoints,
+                             self->p2p_transfer_id++,
+                             payload,
+                             feedback,
+                             user_transfer_reference);
     }
     return out;
 }

libudpard/udpard.h

@@ -329,7 +329,7 @@ struct udpard_tx_t
     /// The globally unique identifier of the local node. Must not change after initialization.
     uint64_t local_uid;
 
-    /// A random-initialized transfer-ID counter for all outgoing P2P transfers.
+    /// A random-initialized transfer-ID counter for all outgoing P2P transfers. Must not be changed by the application.
     uint64_t p2p_transfer_id;
 
     /// The maximum number of Cyphal transfer payload bytes per UDP datagram.
@@ -392,10 +392,6 @@ bool udpard_tx_new(udpard_tx_t* const              self,
 /// transmission queue at the appropriate position. The transfer payload will be copied into the transmission queue
 /// so that the lifetime of the datagrams is not related to the lifetime of the input payload buffer.
 ///
-/// The topic hash is not defined for P2P transfers since there are no topics involved; in P2P, this parameter
-/// is used to pass the destination node's UID instead. Setting it incorrectly will cause the destination node
-/// to reject the transfer as misaddressed.
-///
 /// The transfer_id parameter is used to populate the transfer_id field of the generated Cyphal/UDP frames.
 /// The caller shall increment the transfer-ID counter after each successful invocation of this function
 /// per redundant interface; the same transfer published over redundant interfaces shall have the same transfer-ID.
@@ -435,13 +431,24 @@ uint32_t udpard_tx_push(udpard_tx_t* const      self,
                         const udpard_us_t       now,
                         const udpard_us_t       deadline,
                         const udpard_prio_t     priority,
-                        const uint64_t          topic_hash, // For P2P transfers, this is the destination's UID.
+                        const uint64_t          topic_hash,
                         const udpard_udpip_ep_t remote_ep[UDPARD_IFACE_COUNT_MAX], // May be invalid for some ifaces.
                         const uint64_t          transfer_id,
                         const udpard_bytes_t    payload,
                         void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t), // NULL if best-effort.
                         void* const user_transfer_reference);
 
+/// Specialization for P2P transfers. The semantics are identical to udpard_tx_push().
+/// The transfer-ID will be provided by the library based on the udpard_tx_t::p2p_transfer_id counter.
+uint32_t udpard_tx_push_p2p(udpard_tx_t* const    self,
+                            const udpard_us_t     now,
+                            const udpard_us_t     deadline,
+                            const udpard_prio_t   priority,
+                            const udpard_remote_t remote, // Endpoints may be invalid for some ifaces.
+                            const udpard_bytes_t  payload,
+                            void (*const feedback)(udpard_tx_t*, udpard_tx_feedback_t), // NULL if best-effort.
+                            void* const user_transfer_reference);
+
 /// This should be invoked whenever the socket/NIC of this queue becomes ready to accept new datagrams for transmission.
 /// It is fine to also invoke it periodically unconditionally to drive the transmission process.
 /// Internally, the function will query the scheduler for the next frame to be transmitted and will attempt