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

This reverts commit 6ac82bc.

Author: pavel-kirienko

libudpard/udpard.c

@@ -1656,6 +1656,8 @@ 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.
@@ -1672,46 +1674,30 @@ typedef struct
     udpard_tree_t* fragments;
 } rx_slot_t;
 
-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)
+static void rx_slot_reset(rx_slot_t* const slot, const udpard_mem_t fragment_memory)
 {
-    UDPARD_ASSERT(slot != NULL);
     udpard_fragment_free_all((udpard_fragment_t*)slot->fragments, udpard_make_deleter(fragment_memory));
-    mem_free(slot_memory, sizeof(rx_slot_t), slot);
+    slot->fragments      = NULL;
+    slot->busy           = false;
+    slot->covered_prefix = 0U;
+    slot->crc_end        = 0U;
+    slot->crc            = CRC_INITIAL;
 }
 
-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 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)) {
+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;
         slot->transfer_id = frame->meta.transfer_id;
         slot->ts_min      = ts;
         slot->ts_max      = ts;
@@ -1723,7 +1709,8 @@ static rx_slot_update_result_t 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);
-        return rx_slot_reset;
+        rx_slot_reset(slot, fragment_memory);
+        return false;
     }
     const rx_fragment_tree_update_result_t tree_res = rx_fragment_tree_update(&slot->fragments,
                                                                               fragment_memory,
@@ -1746,12 +1733,14 @@ static rx_slot_update_result_t rx_slot_update(rx_slot_t* const       slot,
     }
     if (tree_res == rx_fragment_tree_done) {
         if (rx_fragment_tree_finalize(slot->fragments, slot->crc)) {
-            return rx_slot_done;
+            slot->busy = false;
+            done       = true;
+        } else {
+            ++*errors_transfer_malformed;
+            rx_slot_reset(slot, fragment_memory);
         }
-        ++*errors_transfer_malformed;
-        return rx_slot_reset;
     }
-    return rx_slot_not_done;
+    return done;
 }
 
 // ---------------------------------------------  SESSION & PORT  ---------------------------------------------
@@ -1763,6 +1752,8 @@ 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;
@@ -1774,9 +1765,10 @@ typedef struct rx_session_t
 
     bool initialized; ///< Set after the first frame is seen.
 
-    udpard_rx_port_t* port;
-
-    rx_slot_t* slots[RX_SLOT_COUNT];
+    // 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];
 } rx_session_t;
 
 /// The reassembly strategy is composed once at initialization time by choosing a vtable with the desired behavior.
@@ -1829,7 +1821,8 @@ 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] = NULL;
+            out->slots[i].fragments = NULL;
+            rx_slot_reset(&out->slots[i], args->owner->memory.fragment);
         }
         out->remote.uid       = args->remote_uid;
         out->port             = args->owner;
@@ -1845,9 +1838,7 @@ 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++) {
-        if (self->slots[i] != NULL) {
-            rx_slot_destroy(self->slots[i], self->port->memory.fragment, self->port->memory.session);
-        }
+        rx_slot_reset(&self->slots[i], self->port->memory.fragment);
     }
     cavl2_remove(&self->port->index_session_by_remote_uid, &self->index_remote_uid);
     delist(sessions_by_animation, &self->list_by_animation);
@@ -1875,44 +1866,46 @@ 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_destroy(slot, self->port->memory.fragment, self->port->memory.session);
+    rx_slot_reset(slot, self->port->memory.fragment);
 }
 
-/// Finds an existing in-progress slot with the specified transfer-ID, or allocates a new one. Returns NULL of OOM.
+/// 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.
 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] != NULL) && (self->slots[i]->transfer_id == transfer_id)) {
-            return self->slots[i];
+        if (self->slots[i].busy && (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] != NULL) && (ts >= (self->slots[i]->ts_max + SESSION_LIFETIME))) {
-            rx_slot_destroy(self->slots[i], self->port->memory.fragment, self->port->memory.session);
+        if (self->slots[i].busy && (ts >= (self->slots[i].ts_max + SESSION_LIFETIME))) {
+            rx_slot_reset(&self->slots[i], self->port->memory.fragment);
         }
     }
     // 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] == NULL) {
-            self->slots[i] = rx_slot_new(self->port->memory.session); // may fail
-            return self->slots[i];
+        if (!self->slots[i].busy) {
+            return &self->slots[i];
         }
     }
     // All slots are currently occupied; find the oldest slot to sacrifice.
-    size_t oldest_index = 0;
+    rx_slot_t*  slot      = NULL;
+    udpard_us_t oldest_ts = HEAT_DEATH;
     for (size_t i = 0; i < RX_SLOT_COUNT; 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(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((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_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];
+    rx_slot_reset(slot, self->port->memory.fragment);
+    UDPARD_ASSERT((slot != NULL) && !slot->busy);
+    return slot;
 }
 
 static void rx_session_update(rx_session_t* const     self,
@@ -1947,24 +1940,21 @@ 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
-        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);
+        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);
             }
+            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,7 +28,6 @@
 ///   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:
@@ -624,8 +623,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 and rx_slot_t.
-    /// Instances are fixed-size and similar in size, so a trivial zero-fragmentation block allocator is sufficient.
+    /// Provides memory for rx_session_t described below.
+    /// Each instance is fixed-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.