rx: allocate slots dynamically to reduce the average case session heap footprint

Author: pavel-kirienko

libudpard/udpard.c

@@ -1656,8 +1656,6 @@ static bool rx_fragment_tree_finalize(udpard_tree_t* const root, const uint32_t
 /// The redundant interfaces may use distinct MTU, which requires special fragment tree handling.
 typedef struct
 {
-    bool busy;
-
     uint64_t transfer_id; ///< Which transfer we're reassembling here.
 
     udpard_us_t ts_min; ///< Earliest frame timestamp, aka transfer reception timestamp.
@@ -1674,30 +1672,46 @@ typedef struct
     udpard_tree_t* fragments;
 } rx_slot_t;
 
-static void rx_slot_reset(rx_slot_t* const slot, const udpard_mem_t fragment_memory)
+static rx_slot_t* rx_slot_new(const udpard_mem_t slot_memory)
+{
+    rx_slot_t* const slot = mem_alloc(slot_memory, sizeof(rx_slot_t));
+    if (slot != NULL) {
+        mem_zero(sizeof(*slot), slot);
+        slot->ts_min         = HEAT_DEATH;
+        slot->ts_max         = BIG_BANG;
+        slot->covered_prefix = 0;
+        slot->crc_end        = 0;
+        slot->crc            = CRC_INITIAL;
+        slot->fragments      = NULL;
+    }
+    return slot;
+}
+
+static void rx_slot_destroy(rx_slot_t* const slot, const udpard_mem_t fragment_memory, const udpard_mem_t slot_memory)
 {
+    UDPARD_ASSERT(slot != NULL);
     udpard_fragment_free_all((udpard_fragment_t*)slot->fragments, udpard_make_deleter(fragment_memory));
-    slot->fragments      = NULL;
-    slot->busy           = false;
-    slot->covered_prefix = 0U;
-    slot->crc_end        = 0U;
-    slot->crc            = CRC_INITIAL;
+    mem_free(slot_memory, sizeof(rx_slot_t), slot);
 }
 
+typedef enum
+{
+    rx_slot_not_done,
+    rx_slot_done,
+    rx_slot_reset,
+} rx_slot_update_result_t;
+
 /// The caller will accept the ownership of the fragments iff the result is true.
-static bool rx_slot_update(rx_slot_t* const       slot,
-                           const udpard_us_t      ts,
-                           const udpard_mem_t     fragment_memory,
-                           const udpard_deleter_t payload_deleter,
-                           rx_frame_t* const      frame,
-                           const size_t           extent,
-                           uint64_t* const        errors_oom,
-                           uint64_t* const        errors_transfer_malformed)
-{
-    bool done = false;
-    if (!slot->busy) {
-        rx_slot_reset(slot, fragment_memory);
-        slot->busy        = true;
+static rx_slot_update_result_t rx_slot_update(rx_slot_t* const       slot,
+                                              const udpard_us_t      ts,
+                                              const udpard_mem_t     fragment_memory,
+                                              const udpard_deleter_t payload_deleter,
+                                              rx_frame_t* const      frame,
+                                              const size_t           extent,
+                                              uint64_t* const        errors_oom,
+                                              uint64_t* const        errors_transfer_malformed)
+{
+    if ((slot->ts_min == HEAT_DEATH) && (slot->ts_max == BIG_BANG)) {
         slot->transfer_id = frame->meta.transfer_id;
         slot->ts_min      = ts;
         slot->ts_max      = ts;
@@ -1709,8 +1723,7 @@ static bool rx_slot_update(rx_slot_t* const       slot,
     if ((slot->total_size != frame->meta.transfer_payload_size) || (slot->priority != frame->meta.priority)) {
         ++*errors_transfer_malformed;
         mem_free_payload(payload_deleter, frame->base.origin);
-        rx_slot_reset(slot, fragment_memory);
-        return false;
+        return rx_slot_reset;
     }
     const rx_fragment_tree_update_result_t tree_res = rx_fragment_tree_update(&slot->fragments,
                                                                               fragment_memory,
@@ -1733,14 +1746,12 @@ static bool rx_slot_update(rx_slot_t* const       slot,
     }
     if (tree_res == rx_fragment_tree_done) {
         if (rx_fragment_tree_finalize(slot->fragments, slot->crc)) {
-            slot->busy = false;
-            done       = true;
-        } else {
-            ++*errors_transfer_malformed;
-            rx_slot_reset(slot, fragment_memory);
+            return rx_slot_done;
         }
+        ++*errors_transfer_malformed;
+        return rx_slot_reset;
     }
-    return done;
+    return rx_slot_not_done;
 }
 
 // ---------------------------------------------  SESSION & PORT  ---------------------------------------------
@@ -1752,8 +1763,6 @@ typedef struct rx_session_t
     udpard_tree_t   index_remote_uid; ///< Must be the first member.
     udpard_remote_t remote;           ///< Most recent discovered reverse path for P2P to the sender.
 
-    udpard_rx_port_t* port;
-
     /// LRU last animated list for automatic retirement of stale sessions.
     udpard_listed_t list_by_animation;
     udpard_us_t     last_animated_ts;
@@ -1765,10 +1774,9 @@ typedef struct rx_session_t
 
     bool initialized; ///< Set after the first frame is seen.
 
-    // TODO: Static slots are taking too much space; allocate them dynamically instead.
-    //       Each is <=56 bytes so it fits nicely into a 64-byte o1heap block.
-    //       The slot state enum can be replaced with a simple "done" flag.
-    rx_slot_t slots[RX_SLOT_COUNT];
+    udpard_rx_port_t* port;
+
+    rx_slot_t* slots[RX_SLOT_COUNT];
 } rx_session_t;
 
 /// The reassembly strategy is composed once at initialization time by choosing a vtable with the desired behavior.
@@ -1821,8 +1829,7 @@ static udpard_tree_t* cavl_factory_rx_session_by_remote_uid(void* const user)
         out->index_remote_uid  = (udpard_tree_t){ NULL, { NULL, NULL }, 0 };
         out->list_by_animation = (udpard_listed_t){ NULL, NULL };
         for (size_t i = 0; i < RX_SLOT_COUNT; i++) {
-            out->slots[i].fragments = NULL;
-            rx_slot_reset(&out->slots[i], args->owner->memory.fragment);
+            out->slots[i] = NULL;
         }
         out->remote.uid       = args->remote_uid;
         out->port             = args->owner;
@@ -1838,7 +1845,9 @@ static udpard_tree_t* cavl_factory_rx_session_by_remote_uid(void* const user)
 static void rx_session_free(rx_session_t* const self, udpard_list_t* const sessions_by_animation)
 {
     for (size_t i = 0; i < RX_SLOT_COUNT; i++) {
-        rx_slot_reset(&self->slots[i], self->port->memory.fragment);
+        if (self->slots[i] != NULL) {
+            rx_slot_destroy(self->slots[i], self->port->memory.fragment, self->port->memory.session);
+        }
     }
     cavl2_remove(&self->port->index_session_by_remote_uid, &self->index_remote_uid);
     delist(sessions_by_animation, &self->list_by_animation);
@@ -1866,46 +1875,44 @@ static void rx_session_eject(rx_session_t* const self, udpard_rx_t* const rx, rx
 
     // Finally, reset the slot.
     slot->fragments = NULL; // Transfer ownership to the application.
-    rx_slot_reset(slot, self->port->memory.fragment);
+    rx_slot_destroy(slot, self->port->memory.fragment, self->port->memory.session);
 }
 
-/// Finds an existing in-progress slot with the specified transfer-ID, or allocates a new one.
-/// Allocation always succeeds so the result is never NULL, but it may cancel a stale slot with incomplete transfer.
+/// Finds an existing in-progress slot with the specified transfer-ID, or allocates a new one. Returns NULL of OOM.
 static rx_slot_t* rx_session_get_slot(rx_session_t* const self, const udpard_us_t ts, const uint64_t transfer_id)
 {
     // First, check if one is in progress already; resume it if so.
     for (size_t i = 0; i < RX_SLOT_COUNT; i++) {
-        if (self->slots[i].busy && (self->slots[i].transfer_id == transfer_id)) {
-            return &self->slots[i];
+        if ((self->slots[i] != NULL) && (self->slots[i]->transfer_id == transfer_id)) {
+            return self->slots[i];
         }
     }
     // Use this opportunity to check for timed-out in-progress slots. This may free up a slot for the search below.
     for (size_t i = 0; i < RX_SLOT_COUNT; i++) {
-        if (self->slots[i].busy && (ts >= (self->slots[i].ts_max + SESSION_LIFETIME))) {
-            rx_slot_reset(&self->slots[i], self->port->memory.fragment);
+        if ((self->slots[i] != NULL) && (ts >= (self->slots[i]->ts_max + SESSION_LIFETIME))) {
+            rx_slot_destroy(self->slots[i], self->port->memory.fragment, self->port->memory.session);
         }
     }
     // This appears to be a new transfer, so we will need to allocate a new slot for it.
     for (size_t i = 0; i < RX_SLOT_COUNT; i++) {
-        if (!self->slots[i].busy) {
-            return &self->slots[i];
+        if (self->slots[i] == NULL) {
+            self->slots[i] = rx_slot_new(self->port->memory.session); // may fail
+            return self->slots[i];
         }
     }
     // All slots are currently occupied; find the oldest slot to sacrifice.
-    rx_slot_t*  slot      = NULL;
-    udpard_us_t oldest_ts = HEAT_DEATH;
+    size_t oldest_index = 0;
     for (size_t i = 0; i < RX_SLOT_COUNT; i++) {
-        UDPARD_ASSERT(self->slots[i].busy); // Checked this already.
-        if (self->slots[i].ts_max < oldest_ts) {
-            oldest_ts = self->slots[i].ts_max;
-            slot      = &self->slots[i];
+        UDPARD_ASSERT(self->slots[i] != NULL); // Checked this already.
+        UDPARD_ASSERT(self->slots[oldest_index] != NULL);
+        if (self->slots[i]->ts_max < self->slots[oldest_index]->ts_max) {
+            oldest_index = i;
         }
     }
-    UDPARD_ASSERT((slot != NULL) && slot->busy);
     // It is probably just a stale transfer, so it's a no-brainer to evict it, it's probably dead anyway.
-    rx_slot_reset(slot, self->port->memory.fragment);
-    UDPARD_ASSERT((slot != NULL) && !slot->busy);
-    return slot;
+    rx_slot_destroy(self->slots[oldest_index], self->port->memory.fragment, self->port->memory.session);
+    self->slots[oldest_index] = rx_slot_new(self->port->memory.session); // may fail
+    return self->slots[oldest_index];
 }
 
 static void rx_session_update(rx_session_t* const     self,
@@ -1940,21 +1947,24 @@ static void rx_session_update(rx_session_t* const     self,
     // UNORDERED mode update. There are no other modes now -- there used to be ORDERED in an experimental revision once.
     if (!rx_session_is_transfer_ejected(self, frame->meta.transfer_id)) {
         rx_slot_t* const slot = rx_session_get_slot(self, ts, frame->meta.transfer_id); // new or continuation
-        UDPARD_ASSERT(slot != NULL);
-        UDPARD_ASSERT((!slot->busy) || (slot->transfer_id == frame->meta.transfer_id));
-        const bool done = rx_slot_update(slot,
-                                         ts,
-                                         self->port->memory.fragment,
-                                         payload_deleter,
-                                         frame,
-                                         self->port->extent,
-                                         &rx->errors_oom,
-                                         &rx->errors_transfer_malformed);
-        if (done) {
-            if (frame->meta.kind == frame_msg_reliable) {
-                tx_send_ack(rx, ts, slot->priority, slot->transfer_id, self->remote);
+        if (slot == NULL) {
+            mem_free_payload(payload_deleter, frame->base.origin);
+            rx->errors_oom++;
+        } else {
+            const bool done = rx_slot_update(slot,
+                                             ts,
+                                             self->port->memory.fragment,
+                                             payload_deleter,
+                                             frame,
+                                             self->port->extent,
+                                             &rx->errors_oom,
+                                             &rx->errors_transfer_malformed);
+            if (done) {
+                if (frame->meta.kind == frame_msg_reliable) {
+                    tx_send_ack(rx, ts, slot->priority, slot->transfer_id, self->remote);
+                }
+                rx_session_eject(self, rx, slot);
             }
-            rx_session_eject(self, rx, slot);
         }
     } else { // retransmit ACK if needed
         if ((frame->meta.kind == frame_msg_reliable) && (frame->base.offset == 0U)) {

libudpard/udpard.h

@@ -28,6 +28,7 @@
 ///   The TX pipeline adds a small overhead of sizeof(tx_frame_t).
 /// - sizeof(tx_transfer_t) blocks for the TX pipeline to store outgoing transfer metadata.
 /// - sizeof(rx_session_t) blocks for the RX pipeline to store incoming transfer session metadata.
+/// - sizeof(rx_slot_t) blocks for the RX pipeline to store specific transfer reassembly state.
 /// - sizeof(udpard_fragment_t) blocks for the RX pipeline to store received data fragments.
 ///
 /// Suitable memory allocators may be found here:
@@ -623,8 +624,8 @@ typedef struct udpard_rx_t
 /// simple applications may choose to use the same memory resource implemented via malloc()/free() for all of them.
 typedef struct udpard_rx_mem_resources_t
 {
-    /// Provides memory for rx_session_t described below.
-    /// Each instance is fixed-size, so a trivial zero-fragmentation block allocator is sufficient.
+    /// Provides memory for rx_session_t and rx_slot_t.
+    /// Instances are fixed-size and similar in size, so a trivial zero-fragmentation block allocator is sufficient.
     udpard_mem_t session;
 
     /// The udpard_fragment_t handles are allocated per payload fragment; each contains a pointer to its fragment.