Added support for High Priority forward state

High priority forward state is activated when stack RX that spesific message to forward.
Enabled state only trig new TX operation for Highest priority level and put other messages to queue.
Mode is disabeld when all High priority messages are sended and 5 seconds from last RX messages.
MPX have a new API for indicate enabled or disabled high priory state.
Wi-sun LLC drop Asynch message send request at High priority state.

Author: Juha Heiskanen

source/6LoWPAN/MAC/mpx_api.h

@@ -90,6 +90,14 @@ typedef int8_t mpx_data_cb_register(const mpx_api_t *api, mpx_data_confirm *conf
  */
 typedef void mpx_eui64_purge_request(const mpx_api_t *api, const uint8_t *eui64);
 
+/**
+ * \brief mpx_high_priority_mode_set Enable/Disable MPX high priority mode for TX process
+ * @param api The API which handled the response
+ * @param enable_mode True for enable High Priority mode, False disable
+ *
+ */
+typedef void mpx_high_priority_mode_set(const mpx_api_t *api, bool enable_mode);
+
 /**
  * \brief Struct mpx_api_s defines functions for MPX user for register call backs and send data.
  */
@@ -99,6 +107,7 @@ struct mpx_api_s {
     mpx_header_size_get *mpx_headroom_size_get;     /**< MPX headroom size get in bytes. */
     mpx_data_cb_register *mpx_user_registration;    /**< MPX User cb registration must be call before enable to send or RX data*/
     mpx_eui64_purge_request *mpx_eui64_purge;       /**< MPX Purge EUI-64 related data */
+    mpx_high_priority_mode_set *mpx_priority_mode_set; /**< MPX request to enable / disable High Priority mode */
 };
 
 

source/6LoWPAN/adaptation_interface.c

@@ -101,6 +101,7 @@ typedef struct {
     uint16_t indirect_big_packet_threshold;
     uint16_t max_indirect_big_packets_total;
     uint16_t max_indirect_small_packets_per_child;
+    uint32_t last_rx_high_priority;
     bool fragmenter_active; /*!< Fragmenter state */
     adaptation_etx_update_cb *etx_update_cb;
     mpx_api_t *mpx_api;
@@ -109,6 +110,7 @@ typedef struct {
 } fragmenter_interface_t;
 
 #define LOWPAN_ACTIVE_UNICAST_ONGOING_MAX 10
+#define LOWPAN_HIGH_PRIORITY_STATE_LENGTH 50 //5 seconds 100us ticks
 
 /* Minimum buffer amount and memory size to ensure operation even in out of memory situation
  */
@@ -155,6 +157,7 @@ static bool lowpan_adaptation_indirect_queue_free_message(struct protocol_interf
 static fragmenter_tx_entry_t *lowpan_adaptation_indirect_mac_data_request_active(fragmenter_interface_t *interface_ptr, fragmenter_tx_entry_t *tx_ptr);
 
 static bool lowpan_buffer_tx_allowed(fragmenter_interface_t *interface_ptr, buffer_t *buf);
+static bool lowpan_adaptation_purge_from_mac(struct protocol_interface_info_entry *cur, fragmenter_interface_t *interface_ptr,  uint8_t msduhandle);
 
 static void lowpan_adaptation_etx_update_cb(protocol_interface_info_entry_t *cur, buffer_t *buf, const mcps_data_conf_t *confirm)
 {
@@ -251,13 +254,41 @@ static void lowpan_adaptation_tx_queue_write(fragmenter_interface_t *interface_p
     protocol_stats_update(STATS_AL_TX_QUEUE_SIZE, interface_ptr->directTxQueue_size);
 }
 
+static void lowpan_adaptation_tx_queue_write_to_front(fragmenter_interface_t *interface_ptr, buffer_t *buf)
+{
+    buffer_t *lower_priority_buf = NULL;
+
+    ns_list_foreach(buffer_t, cur, &interface_ptr->directTxQueue) {
+
+        if (cur->priority <= buf->priority) {
+            lower_priority_buf = cur;
+            break;
+        }
+    }
+
+    if (lower_priority_buf) {
+        ns_list_add_before(&interface_ptr->directTxQueue, lower_priority_buf, buf);
+    } else {
+        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);
+}
+
 static buffer_t *lowpan_adaptation_tx_queue_read(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) {
         return NULL;
     }
     ns_list_foreach_safe(buffer_t, buf, &interface_ptr->directTxQueue) {
+
+        if (buf->link_specific.ieee802_15_4.requestAck && interface_ptr->last_rx_high_priority &&  buf->priority < QOS_EXPEDITE_FORWARD) {
+            //Stop reading at this point when Priority is not enough big
+            return NULL;
+        }
+
         if (lowpan_buffer_tx_allowed(interface_ptr, buf)) {
             ns_list_remove(&interface_ptr->directTxQueue, buf);
             interface_ptr->directTxQueue_size--;
@@ -460,6 +491,7 @@ int8_t lowpan_adaptation_interface_reset(int8_t interface_id)
     buffer_free_list(&interface_ptr->directTxQueue);
     interface_ptr->directTxQueue_size = 0;
     interface_ptr->directTxQueue_level = 0;
+    interface_ptr->last_rx_high_priority = 0;
 
     return 0;
 }
@@ -1075,9 +1107,110 @@ static bool lowpan_buffer_tx_allowed(fragmenter_interface_t *interface_ptr, buff
     if (is_unicast && lowpan_adaptation_is_destination_tx_active(&interface_ptr->activeUnicastList, buf)) {
         return false;
     }
+
+    if (is_unicast && interface_ptr->last_rx_high_priority &&  buf->priority < QOS_EXPEDITE_FORWARD) {
+        return false;
+    }
+    return true;
+}
+
+static uint32_t lowpan_adaptation_time_stamp_diff(uint32_t compare_stamp)
+{
+    if (protocol_core_monotonic_time < compare_stamp) {
+        return compare_stamp - protocol_core_monotonic_time;
+    }
+    return protocol_core_monotonic_time - compare_stamp;
+}
+
+static bool lowpan_adaptation_high_priority_state_exit(fragmenter_interface_t *interface_ptr)
+{
+    if (!interface_ptr->last_rx_high_priority || lowpan_adaptation_time_stamp_diff(interface_ptr->last_rx_high_priority) < LOWPAN_HIGH_PRIORITY_STATE_LENGTH) {
+        return false;
+    }
+
+    //Check First buffer_from tx queue
+    buffer_t *buf = ns_list_get_first(&interface_ptr->directTxQueue);
+    if (buf && buf->priority == QOS_EXPEDITE_FORWARD) {
+        //TX queue must not include any
+        return false;
+    }
+
+    //Check If we have a Any active TX process still active
+    ns_list_foreach(fragmenter_tx_entry_t, entry, &interface_ptr->activeUnicastList) {
+        if (entry->buf->priority == QOS_EXPEDITE_FORWARD) {
+            return false;
+        }
+    }
+
+    //Disable High Priority Mode
+    if (interface_ptr->mpx_api) {
+        interface_ptr->mpx_api->mpx_priority_mode_set(interface_ptr->mpx_api, false);
+    }
+    interface_ptr->last_rx_high_priority = 0;
     return true;
 }
 
+static void lowpan_adaptation_high_priority_state_enable(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr)
+{
+
+    if (!interface_ptr->last_rx_high_priority) {
+        // MPX enaled stack must inform MPX to priority enable
+        if (interface_ptr->mpx_api) {
+            interface_ptr->mpx_api->mpx_priority_mode_set(interface_ptr->mpx_api, true);
+        }
+        //Purge Active tx queue's all possible's
+        if (!interface_ptr->fragmenter_active) {
+            //Purge Only When Fragmenter is not active
+            ns_list_foreach_reverse_safe(fragmenter_tx_entry_t, entry, &interface_ptr->activeUnicastList) {
+
+                if (lowpan_adaptation_purge_from_mac(cur, interface_ptr, entry->buf->seq)) {
+                    buffer_t *buf = entry->buf;
+                    ns_list_remove(&interface_ptr->activeUnicastList, entry);
+                    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);
+                }
+            }
+        }
+    }
+
+    //Store timestamp for indicate last RX High Priority message
+    interface_ptr->last_rx_high_priority = protocol_core_monotonic_time ? protocol_core_monotonic_time : 1;
+
+}
+
+
+static void lowpan_adaptation_priority_status_update(protocol_interface_info_entry_t *cur, fragmenter_interface_t *interface_ptr, buffer_priority_t priority)
+{
+    if (priority == QOS_EXPEDITE_FORWARD) {
+        lowpan_adaptation_high_priority_state_enable(cur, interface_ptr);
+    } else {
+        //Let check can we disable possible High Priority state
+        lowpan_adaptation_high_priority_state_exit(interface_ptr);
+    }
+}
+
+void lowpan_adaptation_interface_slow_timer(protocol_interface_info_entry_t *cur)
+{
+    fragmenter_interface_t *interface_ptr = lowpan_adaptation_interface_discover(cur->id);
+    if (!interface_ptr) {
+        return;
+    }
+
+    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);
+        while (buf_from_queue) {
+            lowpan_adaptation_interface_tx(cur, buf_from_queue);
+            buf_from_queue = lowpan_adaptation_tx_queue_read(interface_ptr);
+        }
+        //Update Average QUEUE
+        random_early_detetction_aq_calc(cur->random_early_detection, interface_ptr->directTxQueue_size);
+    }
+}
+
 int8_t lowpan_adaptation_interface_tx(protocol_interface_info_entry_t *cur, buffer_t *buf)
 {
     bool is_room_for_new_message;
@@ -1105,6 +1238,8 @@ int8_t lowpan_adaptation_interface_tx(protocol_interface_info_entry_t *cur, buff
         buffer_priority_set(buf, QOS_HIGH);
     }
 
+    //Update priority status
+    lowpan_adaptation_priority_status_update(cur, interface_ptr, buf->priority);
 
     //Check packet size
     bool fragmented_needed = lowpan_adaptation_request_longer_than_mtu(cur, buf, interface_ptr);
@@ -1409,6 +1544,8 @@ int8_t lowpan_adaptation_interface_tx_confirm(protocol_interface_info_entry_t *c
     }
     // When confirmation is for direct transmission, push all allowed buffers to MAC
     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);
         while (buf_from_queue) {
             lowpan_adaptation_interface_tx(cur, buf_from_queue);

source/6LoWPAN/lowpan_adaptation_interface.h

@@ -62,4 +62,6 @@ int8_t lowpan_adaptation_free_messages_from_queues_by_address(struct protocol_in
 
 int8_t lowpan_adaptation_indirect_queue_params_set(struct protocol_interface_info_entry *cur, uint16_t indirect_big_packet_threshold, uint16_t max_indirect_big_packets_total, uint16_t max_indirect_small_packets_per_child);
 
+void lowpan_adaptation_interface_slow_timer(struct protocol_interface_info_entry *cur);
+
 #endif /* LOWPAN_ADAPTATION_INTERFACE_H_ */

source/6LoWPAN/ws/ws_llc_data_service.c

@@ -135,6 +135,7 @@ typedef struct {
     ws_neighbor_info_request        *ws_neighbor_info_request_cb;   /**< LLC Neighbour discover API*/
     uint8_t                         ws_enhanced_response_elements[ENHANCED_FRAME_RESPONSE];
     ns_ie_iovec_t                   ws_header_vector;
+    bool                            high_priority_mode;
     protocol_interface_info_entry_t *interface_ptr;                 /**< List link entry */
 } llc_data_base_t;
 
@@ -1343,6 +1344,15 @@ static uint8_t ws_llc_mpx_data_purge_request(const mpx_api_t *api, struct mcps_p
     return purge_status;
 }
 
+static void wc_llc_mpx_priority_set_request(const mpx_api_t *api, bool enable_mode)
+{
+    llc_data_base_t *base = ws_llc_discover_by_mpx(api);
+    if (!base) {
+        return;
+    }
+    base->high_priority_mode = enable_mode;
+}
+
 static void ws_llc_mpx_init(mpx_class_t *mpx_class)
 {
     //Init Mbed Class and API
@@ -1353,6 +1363,7 @@ static void ws_llc_mpx_init(mpx_class_t *mpx_class)
     mpx_class->mpx_api.mpx_data_request = &ws_llc_mpx_data_request;
     mpx_class->mpx_api.mpx_data_purge = &ws_llc_mpx_data_purge_request;
     mpx_class->mpx_api.mpx_eui64_purge = &ws_llc_mpx_eui64_purge_request;
+    mpx_class->mpx_api.mpx_priority_mode_set = &wc_llc_mpx_priority_set_request;
 }
 
 static void ws_llc_clean(llc_data_base_t *base)
@@ -1375,6 +1386,8 @@ static void ws_llc_clean(llc_data_base_t *base)
     memset(&base->ie_params, 0, sizeof(llc_ie_params_t));
 
     ws_llc_temp_neigh_info_table_reset(base->temp_entries);
+    //Disable High Priority mode
+    base->high_priority_mode = false;
 }
 
 static void ws_llc_temp_neigh_info_table_reset(temp_entriest_t *base)
@@ -1677,6 +1690,11 @@ int8_t ws_llc_asynch_request(struct protocol_interface_info_entry *interface, as
         return -1;
     }
 
+    if (base->high_priority_mode) {
+        //Drop asynch messages at High Priority mode
+        return -1;
+    }
+
 
     //Calculate IE Buffer size
     request->wh_requested_ie_list.fc_ie = false; //Never should not be a part Asynch message

source/NWK_INTERFACE/protocol_core.c

@@ -265,6 +265,7 @@ void core_timer_event_handle(uint16_t ticksUpdate)
                         cur->nwk_wpan_nvm_api->nvm_params_update_cb(cur->nwk_wpan_nvm_api, false);
                     }
                     etx_cache_timer(cur->id, seconds);
+                    lowpan_adaptation_interface_slow_timer(cur);
                 }
             } else if (cur->nwk_id == IF_IPV6) {
                 //Slow Pointer Update

test/nanostack/unittest/stub/adaptation_interface_stub.c

@@ -111,3 +111,8 @@ void lowpan_adaptation_neigh_remove_free_tx_tables(struct protocol_interface_inf
 {
 
 }
+
+void lowpan_adaptation_interface_slow_timer(protocol_interface_info_entry_t *cur)
+{
+
+}