udpard_tx_pending_iface_mask

Author: pavel-kirienko

libudpard/udpard.c

@@ -1301,6 +1301,24 @@ void udpard_tx_poll(udpard_tx_t* const self, const udpard_us_t now, const uint32
     }
 }
 
+uint32_t udpard_tx_pending_iface_mask(const udpard_tx_t* const self)
+{
+    uint32_t mask = 0;
+    if (self != NULL) {
+        // Even though it's constant-time, I still mildly dislike this loop. Shall it become a bottleneck,
+        // we could modify the TX state to keep a mask of pending interfaces updated incrementally.
+        for (size_t i = 0; i < UDPARD_IFACE_COUNT_MAX; i++) {
+            for (size_t p = 0; p < UDPARD_PRIORITY_COUNT; p++) {
+                if (self->queue[i][p].head != NULL) {
+                    mask |= (1U << i);
+                    break;
+                }
+            }
+        }
+    }
+    return mask;
+}
+
 void udpard_tx_refcount_inc(const udpard_bytes_t tx_payload_view)
 {
     if (tx_payload_view.data != NULL) {

libudpard/udpard.h

@@ -540,6 +540,11 @@ uint32_t udpard_tx_push_p2p(udpard_tx_t* const             self,
 /// The function may deallocate memory. The time complexity is logarithmic in the number of enqueued transfers.
 void udpard_tx_poll(udpard_tx_t* const self, const udpard_us_t now, const uint32_t iface_mask);
 
+/// Returns a bitmask of interfaces that have pending transmissions. This is useful for IO multiplexing loops.
+/// Zero indicates that there are no pending transmissions.
+/// Which interfaces are usable is defined by the remote endpoints provided when pushing transfers.
+uint32_t udpard_tx_pending_iface_mask(const udpard_tx_t* const self);
+
 /// When a datagram is ejected and the application opts to keep it, these functions must be used to manage the
 /// datagram buffer lifetime. The datagram will be freed once the reference count reaches zero.
 void udpard_tx_refcount_inc(const udpard_bytes_t tx_payload_view);

tests/src/test_intrusive_tx.c

@@ -375,9 +375,11 @@ static void test_tx_ack_and_scheduler(void)
                                             make_scattered(NULL, 0),
                                             record_feedback,
                                             make_user_context(&fstate)));
+    TEST_ASSERT_EQUAL_UINT32(1U << 0U, udpard_tx_pending_iface_mask(&tx1));
     udpard_rx_t rx = { .tx = &tx1 };
     tx_receive_ack(&rx, 21, 42);
     TEST_ASSERT_EQUAL_size_t(1, fstate.count);
+    TEST_ASSERT_EQUAL_UINT32(0U, udpard_tx_pending_iface_mask(&tx1));
     udpard_tx_free(&tx1);
 
     // Ack suppressed when coverage not improved.
@@ -397,6 +399,7 @@ static void test_tx_ack_and_scheduler(void)
     rx.tx            = &tx2;
     tx_send_ack(&rx, 0, udpard_prio_fast, 7, 8, (udpard_remote_t){ .uid = 9, .endpoints = { make_ep(3) } });
     TEST_ASSERT_EQUAL_UINT64(0, rx.errors_ack_tx);
+    TEST_ASSERT_EQUAL_UINT32(0U, udpard_tx_pending_iface_mask(&tx2));
     udpard_tx_free(&tx2);
 
     // Ack replaced with broader coverage.
@@ -406,6 +409,7 @@ static void test_tx_ack_and_scheduler(void)
     tx_send_ack(&rx, 0, udpard_prio_fast, 9, 9, (udpard_remote_t){ .uid = 11, .endpoints = { make_ep(4) } });
     tx_send_ack(
       &rx, 0, udpard_prio_fast, 9, 9, (udpard_remote_t){ .uid = 11, .endpoints = { make_ep(4), make_ep(5) } });
+    TEST_ASSERT_NOT_EQUAL(0U, udpard_tx_pending_iface_mask(&tx3));
     udpard_tx_free(&tx3);
 
     // Ack push failure with TX present.
@@ -466,6 +470,7 @@ static void test_tx_ack_and_scheduler(void)
     tx5.ack_baseline_timeout = 1;
     tx_promote_staged_transfers(&tx5, 1);
     TEST_ASSERT_NOT_NULL(tx5.queue[0][staged.priority].head);
+    TEST_ASSERT_EQUAL_UINT32(1U << 0U, udpard_tx_pending_iface_mask(&tx5));
 
     // Ejection stops when NIC refuses.
     staged.cursor[0]                   = staged.head[0];
@@ -535,7 +540,7 @@ static void test_tx_stage_if_via_tx_push(void)
     TEST_ASSERT_TRUE(udpard_tx_new(&tx, 30U, 1U, 4U, mem, &vt));
     tx.user                                        = &log;
     tx.ack_baseline_timeout                        = 10;
-    udpard_udpip_ep_t dest[UDPARD_IFACE_COUNT_MAX] = { make_ep(1), { 0 } };
+    udpard_udpip_ep_t dest[UDPARD_IFACE_COUNT_MAX] = { make_ep(1), make_ep(2), { 0 } };
 
     TEST_ASSERT_GREATER_THAN_UINT32(0,
                                     udpard_tx_push(&tx,
@@ -548,14 +553,18 @@ static void test_tx_stage_if_via_tx_push(void)
                                                    make_scattered(NULL, 0),
                                                    record_feedback,
                                                    make_user_context(&fb)));
+    TEST_ASSERT_EQUAL_UINT32((1U << 0U) | (1U << 1U), udpard_tx_pending_iface_mask(&tx));
 
     udpard_tx_poll(&tx, 0, UDPARD_IFACE_MASK_ALL);
     udpard_tx_poll(&tx, 160, UDPARD_IFACE_MASK_ALL);
     udpard_tx_poll(&tx, 400, UDPARD_IFACE_MASK_ALL);
+    TEST_ASSERT_EQUAL_UINT32(0U, udpard_tx_pending_iface_mask(&tx));
 
-    TEST_ASSERT_EQUAL_size_t(2, log.count);
+    TEST_ASSERT_EQUAL_size_t(4, log.count);
     TEST_ASSERT_EQUAL(0, log.when[0]);
-    TEST_ASSERT_EQUAL(160, log.when[1]);
+    TEST_ASSERT_EQUAL(0, log.when[1]);
+    TEST_ASSERT_EQUAL(160, log.when[2]);
+    TEST_ASSERT_EQUAL(160, log.when[3]);
     TEST_ASSERT_NULL(tx.index_staged);
     udpard_tx_free(&tx);
     instrumented_allocator_reset(&alloc);