Implement Message Queue Limit to Prevent Memory Exhaustion in Embedded Devices

api/oc_network_events.c

@@ -45,6 +45,37 @@ static bool g_interface_up;
 static bool g_interface_down;
 #endif /* OC_NETWORK_MONITOR */
 
+#ifdef OC_DYNAMIC_ALLOCATION
+static size_t
+get_events_queue_length(size_t device, oc_list_t events)
+{
+  size_t msg_count = 0;
+  for (oc_message_t *msg = (oc_message_t *)oc_list_head(events); msg != NULL;
+       msg = msg->next) {
+    if (msg->endpoint.device == device) {
+      ++msg_count;
+    }
+  }
+  return msg_count;
+}
+
+static void
+send_wakeup_signal(oc_list_t events)
+{
+  size_t msg_total = oc_list_length(events);
+
+  int deviceId = 0;
+  while (msg_total >= OC_DEVICE_MAX_NUM_CONCURRENT_REQUESTS) {
+    size_t msg_device_count = get_events_queue_length(deviceId, events);
+    if (msg_device_count >= OC_DEVICE_MAX_NUM_CONCURRENT_REQUESTS) {
+      oc_connectivity_wakeup(deviceId);
+    }
+    msg_total -= msg_device_count;
+    ++deviceId; // advance to the next device
+  }
+}
+#endif /* OC_DYNAMIC_ALLOCATION */
+
 static void
 oc_process_network_event(void)
 {
@@ -65,6 +96,11 @@ oc_process_network_event(void)
 #endif /* OC_NETWORK_MONITOR */
   oc_network_event_handler_mutex_unlock();
 
+#ifdef OC_DYNAMIC_ALLOCATION
+  // send a wake-up signal in case the queue might reach the limit for a device
+  send_wakeup_signal(network_events);
+#endif /* OC_DYNAMIC_ALLOCATION */
+
 #ifdef OC_HAS_FEATURE_TCP_ASYNC_CONNECT
   oc_tcp_on_connect_event_t *event =
     (oc_tcp_on_connect_event_t *)oc_list_pop(network_tcp_connect_events);
@@ -153,10 +189,10 @@ oc_network_tcp_connect_event(oc_tcp_on_connect_event_t *event)
 }
 #endif /* OC_HAS_FEATURE_TCP_ASYNC_CONNECT */
 
-int
+size_t
 oc_network_drop_receive_events(const oc_endpoint_t *endpoint)
 {
-  int dropped = 0;
+  size_t dropped = 0;
   oc_network_event_handler_mutex_lock();
   for (oc_message_t *message = (oc_message_t *)oc_list_head(g_network_events);
        message != NULL;) {
@@ -178,6 +214,15 @@ oc_network_drop_receive_events(const oc_endpoint_t *endpoint)
     }
     message = next;
   }
+
+#ifdef OC_DYNAMIC_ALLOCATION
+  if (get_events_queue_length(endpoint->device, g_network_events) + dropped >=
+      OC_DEVICE_MAX_NUM_CONCURRENT_REQUESTS) {
+    // send a wake-up signal in case the queue for the device was full
+    oc_connectivity_wakeup(endpoint->device);
+  }
+#endif /* OC_DYNAMIC_ALLOCATION */
+
   oc_network_event_handler_mutex_unlock();
   return dropped;
 }
@@ -204,3 +249,14 @@ oc_network_interface_event(oc_interface_event_t event)
   _oc_signal_event_loop();
 }
 #endif /* OC_NETWORK_MONITOR */
+
+#ifdef OC_DYNAMIC_ALLOCATION
+size_t
+oc_network_get_event_queue_length(size_t device)
+{
+  oc_network_event_handler_mutex_lock();
+  size_t msg_count = get_events_queue_length(device, g_network_events);
+  oc_network_event_handler_mutex_unlock();
+  return msg_count;
+}
+#endif /* OC_DYNAMIC_ALLOCATION */

api/oc_network_events_internal.h

@@ -58,7 +58,8 @@ void oc_network_tcp_connect_event(oc_tcp_on_connect_event_t *event)
  * @param endpoint the endpoint (cannot be NULL)
  * @return number of events dropped
  */
-int oc_network_drop_receive_events(const oc_endpoint_t *endpoint) OC_NONNULL();
+size_t oc_network_drop_receive_events(const oc_endpoint_t *endpoint)
+  OC_NONNULL();
 
 #ifdef OC_NETWORK_MONITOR
 /**
@@ -78,6 +79,14 @@ typedef struct oc_network_interface_cb
 void oc_network_interface_event(oc_interface_event_t event);
 #endif /* OC_NETWORK_MONITOR */
 
+/**
+ * @brief Returns the network event queue length for the device
+ *
+ * @param device valid device index
+ * @return number of events in the queue
+ */
+size_t oc_network_get_event_queue_length(size_t device);
+
 #ifdef __cplusplus
 }
 #endif