Adaptation layer to remove oldest packet first

When running out of memory or reducing traffic adaptation
will remove the oldest packets first instead of newest

Also it needs to first remove all Lower priority packets before
remove the next level of priorities

Author: Mika Tervonen

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features
 * Filter RPL parents based on DEVICE_MIN_SENS configuration with Threshold and Hysteresis.
 * Filter EAPOL parents based on DEVICE_MIN_SENS configuration with Threshold and Hysteresis to prevent EAPOL failures caused by bad signal levels
+* Adaptation layer removes oldest packets first when stack needs to release memory or reduce traffic
 
 ### Changes
 * 

source/6LoWPAN/adaptation_interface.c

@@ -128,8 +128,8 @@ static NS_LIST_DEFINE(fragmenter_interface_list, fragmenter_interface_t, link);
 static fragmenter_interface_t *lowpan_adaptation_interface_discover(int8_t interfaceId);
 
 /* Interface direct message pending queue functions */
-static void lowpan_adaptation_tx_queue_write(fragmenter_interface_t *interface_ptr, buffer_t *buf);
-static buffer_t *lowpan_adaptation_tx_queue_read(fragmenter_interface_t *interface_ptr);
+static void lowpan_adaptation_tx_queue_write(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr, buffer_t *buf);
+static buffer_t *lowpan_adaptation_tx_queue_read(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr);
 
 /* Data direction and message length validation */
 static bool lowpan_adaptation_indirect_data_request(mac_neighbor_table_entry_t *mle_entry);
@@ -223,8 +223,11 @@ static struct protocol_interface_info_entry *lowpan_adaptation_network_interface
 }
 
 
-static void lowpan_adaptation_tx_queue_level_update(fragmenter_interface_t *interface_ptr)
+static void lowpan_adaptation_tx_queue_level_update(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr)
 {
+    random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
+    protocol_stats_update(STATS_AL_TX_QUEUE_SIZE, interface_ptr->directTxQueue_size);
+
     if (interface_ptr->directTxQueue_size == interface_ptr->directTxQueue_level + ADAPTION_DIRECT_TX_QUEUE_SIZE_THRESHOLD_TRACE ||
             interface_ptr->directTxQueue_size == interface_ptr->directTxQueue_level - ADAPTION_DIRECT_TX_QUEUE_SIZE_THRESHOLD_TRACE) {
         interface_ptr->directTxQueue_level = interface_ptr->directTxQueue_size;
@@ -233,14 +236,14 @@ static void lowpan_adaptation_tx_queue_level_update(fragmenter_interface_t *inte
 }
 
 
-static void lowpan_adaptation_tx_queue_write(fragmenter_interface_t *interface_ptr, buffer_t *buf)
+static void lowpan_adaptation_tx_queue_write(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr, buffer_t *buf)
 {
     buffer_t *lower_priority_buf = NULL;
 
-    ns_list_foreach(buffer_t, cur, &interface_ptr->directTxQueue) {
+    ns_list_foreach(buffer_t, entry, &interface_ptr->directTxQueue) {
 
-        if (cur->priority < buf->priority) {
-            lower_priority_buf = cur;
+        if (entry->priority < buf->priority) {
+            lower_priority_buf = entry;
             break;
         }
     }
@@ -251,18 +254,17 @@ static void lowpan_adaptation_tx_queue_write(fragmenter_interface_t *interface_p
         ns_list_add_to_end(&interface_ptr->directTxQueue, buf);
     }
     interface_ptr->directTxQueue_size++;
-    lowpan_adaptation_tx_queue_level_update(interface_ptr);
-    protocol_stats_update(STATS_AL_TX_QUEUE_SIZE, interface_ptr->directTxQueue_size);
+    lowpan_adaptation_tx_queue_level_update(cur, interface_ptr);
 }
 
-static void lowpan_adaptation_tx_queue_write_to_front(fragmenter_interface_t *interface_ptr, buffer_t *buf)
+static void lowpan_adaptation_tx_queue_write_to_front(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr, buffer_t *buf)
 {
     buffer_t *lower_priority_buf = NULL;
 
-    ns_list_foreach(buffer_t, cur, &interface_ptr->directTxQueue) {
+    ns_list_foreach(buffer_t, entry, &interface_ptr->directTxQueue) {
 
-        if (cur->priority <= buf->priority) {
-            lower_priority_buf = cur;
+        if (entry->priority <= buf->priority) {
+            lower_priority_buf = entry;
             break;
         }
     }
@@ -273,11 +275,10 @@ static void lowpan_adaptation_tx_queue_write_to_front(fragmenter_interface_t *in
         ns_list_add_to_end(&interface_ptr->directTxQueue, buf);
     }
     interface_ptr->directTxQueue_size++;
-    lowpan_adaptation_tx_queue_level_update(interface_ptr);
-    protocol_stats_update(STATS_AL_TX_QUEUE_SIZE, interface_ptr->directTxQueue_size);
+    lowpan_adaptation_tx_queue_level_update(cur, interface_ptr);
 }
 
-static buffer_t *lowpan_adaptation_tx_queue_read(fragmenter_interface_t *interface_ptr)
+static buffer_t *lowpan_adaptation_tx_queue_read(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr)
 {
     // Currently this function is called only when data confirm is received for previously sent packet.
     if (!interface_ptr->directTxQueue_size) {
@@ -293,8 +294,7 @@ static buffer_t *lowpan_adaptation_tx_queue_read(fragmenter_interface_t *interfa
         if (lowpan_buffer_tx_allowed(interface_ptr, buf)) {
             ns_list_remove(&interface_ptr->directTxQueue, buf);
             interface_ptr->directTxQueue_size--;
-            lowpan_adaptation_tx_queue_level_update(interface_ptr);
-            protocol_stats_update(STATS_AL_TX_QUEUE_SIZE, interface_ptr->directTxQueue_size);
+            lowpan_adaptation_tx_queue_level_update(cur, interface_ptr);
             return buf;
         }
     }
@@ -561,12 +561,24 @@ void lowpan_adaptation_free_heap(bool full_gc)
         lowpan_adaptation_free_low_priority_packets(interface_ptr->interface_id, priority, amount);
     }
 }
+buffer_t *lowpan_adaptation_get_oldest_packet(fragmenter_interface_t *interface_ptr, buffer_priority_t priority)
+{
+    ns_list_foreach(buffer_t, entry, &interface_ptr->directTxQueue) {
+        if (entry->priority == priority) {
+            // Only Higher priority packets left no need to go through list anymore
+            return entry;
+        }
+    }
+    return NULL;
+
+}
 
 int8_t lowpan_adaptation_free_low_priority_packets(int8_t interface_id, buffer_priority_t max_priority, uint32_t requested_amount)
 {
     fragmenter_interface_t *interface_ptr = lowpan_adaptation_interface_discover(interface_id);
+    protocol_interface_info_entry_t *cur = protocol_stack_interface_info_get_by_id(interface_id);
 
-    if (!interface_ptr) {
+    if (!interface_ptr || !cur) {
         return -1;
     }
     uint32_t adaptation_memory = 0;
@@ -592,21 +604,36 @@ int8_t lowpan_adaptation_free_low_priority_packets(int8_t interface_id, buffer_p
         requested_amount = adaptation_memory - LOWPAN_MEM_LIMIT_MIN_MEMORY;
     }
 
-    //Only remove last entries from TX queue with low priority
-    ns_list_foreach_reverse_safe(buffer_t, entry, &interface_ptr->directTxQueue) {
-        if (entry->priority <= max_priority) {
-            memory_freed += sizeof(buffer_t) + entry->size;
-            packets_freed++;
-            ns_list_remove(&interface_ptr->directTxQueue, entry);
-            interface_ptr->directTxQueue_size--;
-            lowpan_adaptation_tx_queue_level_update(interface_ptr);
-            socket_tx_buffer_event_and_free(entry, SOCKET_TX_FAIL);
-        }
-        if (memory_freed > requested_amount) {
-            // Enough memory freed
+    /* Order of packets is
+     * priority 4 oldest to newest
+     * priority 3 oldest to newest
+     * priority 2 oldest to newest
+     * priority 1 oldest to newest
+     * priority 0 oldest to newest
+     * So we search oldest for lowest priority and delete that until that priority is empty
+     * and then start deleting one higher priority packets from oldest first
+     */
+
+    buffer_priority_t priority = QOS_NORMAL;
+    do {
+        buffer_t *entry = lowpan_adaptation_get_oldest_packet(interface_ptr, priority);
+        if (!entry) {
+            if (priority < max_priority) {
+                priority++;
+                continue;
+            }
+            // No more packets available
             break;
         }
-    }
+        // This packet can be deleted
+        memory_freed += sizeof(buffer_t) + entry->size;
+        packets_freed++;
+        ns_list_remove(&interface_ptr->directTxQueue, entry);
+        interface_ptr->directTxQueue_size--;
+        lowpan_adaptation_tx_queue_level_update(cur, interface_ptr);
+        socket_tx_buffer_event_and_free(entry, SOCKET_TX_FAIL);
+    } while (memory_freed < requested_amount);
+
     tr_info("Adaptation Free low priority packets memory: %" PRIi32 " queue: %d deallocated %" PRIi32 " bytes, %d packets, %" PRIi32 " requested", adaptation_memory, adaptation_packets, memory_freed, packets_freed, requested_amount);
     return 0;
 }
@@ -1210,8 +1237,7 @@ static void lowpan_adaptation_high_priority_state_enable(protocol_interface_info
                     interface_ptr->activeTxList_size--;
                     ns_dyn_mem_free(entry);
                     //Add message to tx queue front based on priority. Now same priority at buf is prioritised at order
-                    lowpan_adaptation_tx_queue_write_to_front(interface_ptr, buf);
-                    random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
+                    lowpan_adaptation_tx_queue_write_to_front(cur, interface_ptr, buf);
                 }
             }
         }
@@ -1260,13 +1286,11 @@ void lowpan_adaptation_interface_slow_timer(protocol_interface_info_entry_t *cur
 
     if (lowpan_adaptation_high_priority_state_exit(interface_ptr)) {
         //Activate Packets from TX queue
-        buffer_t *buf_from_queue = lowpan_adaptation_tx_queue_read(interface_ptr);
+        buffer_t *buf_from_queue = lowpan_adaptation_tx_queue_read(cur, interface_ptr);
         while (buf_from_queue) {
             lowpan_adaptation_interface_tx(cur, buf_from_queue);
-            buf_from_queue = lowpan_adaptation_tx_queue_read(interface_ptr);
+            buf_from_queue = lowpan_adaptation_tx_queue_read(cur, interface_ptr);
         }
-        //Update Average QUEUE
-        random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
     }
 }
 
@@ -1319,17 +1343,17 @@ int8_t lowpan_adaptation_interface_tx(protocol_interface_info_entry_t *cur, buff
 
     if (!lowpan_buffer_tx_allowed(interface_ptr, buf)) {
 
-        if (buf->priority == QOS_NORMAL) {
-
-            if (random_early_detection_congestion_check(cur->random_early_detection)) {
-                random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
+        if (random_early_detection_congestion_check(cur->random_early_detection)) {
+            // If we need to drop packet we drop oldest normal Priority packet.
+            buffer_t *dropped = lowpan_adaptation_get_oldest_packet(interface_ptr, QOS_NORMAL);
+            if (dropped) {
+                ns_list_remove(&interface_ptr->directTxQueue, dropped);
+                interface_ptr->directTxQueue_size--;
+                socket_tx_buffer_event_and_free(dropped, SOCKET_TX_FAIL);
                 protocol_stats_update(STATS_AL_TX_CONGESTION_DROP, 1);
-                goto tx_error_handler;
             }
         }
-
-        lowpan_adaptation_tx_queue_write(interface_ptr, buf);
-        random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
+        lowpan_adaptation_tx_queue_write(cur, interface_ptr, buf);
         return 0;
     }
 
@@ -1605,13 +1629,11 @@ int8_t lowpan_adaptation_interface_tx_confirm(protocol_interface_info_entry_t *c
     if (active_direct_confirm == true) {
         //Check Possibility for exit from High Priority state
         lowpan_adaptation_high_priority_state_exit(interface_ptr);
-        buffer_t *buf_from_queue = lowpan_adaptation_tx_queue_read(interface_ptr);
+        buffer_t *buf_from_queue = lowpan_adaptation_tx_queue_read(cur, interface_ptr);
         while (buf_from_queue) {
             lowpan_adaptation_interface_tx(cur, buf_from_queue);
-            buf_from_queue = lowpan_adaptation_tx_queue_read(interface_ptr);
+            buf_from_queue = lowpan_adaptation_tx_queue_read(cur, interface_ptr);
         }
-        //Update Average QUEUE
-        random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
     }
     return 0;
 }
@@ -1815,8 +1837,7 @@ int8_t lowpan_adaptation_free_messages_from_queues_by_address(struct protocol_in
             ns_list_remove(&interface_ptr->directTxQueue, entry);
             interface_ptr->directTxQueue_size--;
             //Update Average QUEUE
-            random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
-            lowpan_adaptation_tx_queue_level_update(interface_ptr);
+            lowpan_adaptation_tx_queue_level_update(cur, interface_ptr);
             socket_tx_buffer_event_and_free(entry, SOCKET_TX_FAIL);
         }
     }