[SYS] Improve central queue stability (#1796)

* [SYS] Improve central queue stability

Improve central queue stability and add metrics

* Change partition for esp32dev-pilight-somfy-cc1101

to avoid program size too big error message with esp32dev-pilight-somfy-cc1101

* Only check for gateway updates at the start

to avoid consuming extra memory during operations
Publish the information about the available release with the retain flag

environments.ini

@@ -185,6 +185,7 @@ custom_description = Gateway using Somfy Remote library, need CC1101
 [env:esp32dev-pilight-somfy-cc1101]
 platform = ${com.esp32_platform}
 board = esp32dev
+board_build.partitions = min_spiffs.csv
 lib_deps =
   ${com-esp32.lib_deps}
   ${libraries.esppilight}

main/ZgatewayBT.ino

@@ -662,8 +662,8 @@ void procBLETask(void* pvParameters) {
  */
 void BLEscan() {
   // Don't start the next scan until processing of previous results is complete.
-  while (uxQueueMessagesWaiting(BLEQueue) || jsonQueue.size() != 0) {
+  while (uxQueueMessagesWaiting(BLEQueue) || queueLength != 0) { // the criteria on queueLength could be adjusted to parallelize the scan and the queue processing
-    yield();
+    delay(1); // Wait for queue to empty, a yield here instead of the delay cause the WDT to trigger
   }
   Log.notice(F("Scan begin" CR));
   BLEScan* pBLEScan = BLEDevice::getScan();
@@ -806,9 +806,8 @@ void coreTask(void* pvParameters) {
           delay(waitms = interval > 100 ? 100 : interval); // 100ms
         }
       }
-    } else {
-      Log.trace(F("BLE core task canceled by processLock" CR));
     }
+    delay(1);
   }
 }
 
@@ -872,7 +871,7 @@ void setupBTTasksAndBLE() {
   xTaskCreatePinnedToCore(
       procBLETask, /* Function to implement the task */
       "procBLETask", /* Name of the task */
-      8000, /* Stack size in bytes */
+      8500, /* Stack size in bytes */
       NULL, /* Task input parameter */
       2, /* Priority of the task (set higher than core task) */
       &xProcBLETaskHandle, /* Task handle. */
@@ -1031,7 +1030,7 @@ void launchBTDiscovery(bool overrideDiscovery) {
                                 0, "", "", false, "",
                                 model.c_str(), brand.c_str(), model_id.c_str(), macWOdots.c_str(), false,
                                 stateClassMeasurement, nullptr, nullptr, "[\"lb\",\"kg\",\"jin\"]");
-              } else if (strcmp(prop.key().c_str(), "device") != 0 || strcmp(prop.key().c_str(), "mac") != 0) { // Exception on device and mac as these ones are not sensors
+              } else if (strcmp(prop.key().c_str(), "device") != 0 && strcmp(prop.key().c_str(), "mac") != 0) { // Exception on device and mac as these ones are not sensors
                 createDiscovery("sensor",
                                 discovery_topic.c_str(), entity_name.c_str(), unique_id.c_str(),
                                 will_Topic, prop.value()["name"], value_template.c_str(),
@@ -1138,7 +1137,7 @@ void PublishDeviceData(JsonObject& BLEdata) {
 }
 
 void process_bledata(JsonObject& BLEdata) {
-  yield();
+  yield(); // Necessary to let the loop run in case of connectivity issues
   if (!BLEdata.containsKey("id")) {
     Log.error(F("No mac address in the payload" CR));
     return;

main/main.ino

@@ -62,8 +62,12 @@ unsigned long timer_sys_checks = 0;
 #  define ARDUINOJSON_ENABLE_STD_STRING 1
 
 #  include <queue>
+int queueLength = 0;
+int queueLengthSum = 0;
+int blockedMessages = 0;
+int maxQueueLength = 0;
 #  ifndef QueueSize
-#    define QueueSize 32
+#    define QueueSize 18
 #  endif
 
 #endif
@@ -88,7 +92,6 @@ struct JsonBundle {
 };
 
 std::queue<JsonBundle> jsonQueue;
-int queueLengthSum = 0;
 #endif
 
 StaticJsonDocument<JSON_MSG_BUFFER> modulesBuffer;
@@ -264,7 +267,7 @@ static String ota_server_cert = "";
 #  include <nvs.h>
 #  include <nvs_flash.h>
 
-bool ProcessLock = false; // Process lock when we want to use a critical function like OTA for example
+bool ProcessLock = true; // Process lock when we want to use a critical function like OTA for example, at start to true so as to wait for critical functions to be performed before BLE start
 #  if !defined(NO_INT_TEMP_READING)
 // ESP32 internal temperature reading
 #    include <stdio.h>
@@ -426,11 +429,16 @@ void pubMainCore(JsonObject& data) {
 // Add a document to the queue
 void enqueueJsonObject(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc) {
   if (jsonDoc.size() == 0) {
-    Log.error(F("Empty JSON, not enqueued" CR));
+    Log.error(F("Empty JSON, skipping" CR));
+    return;
+  }
+#if defined(ESP8266) || defined(ESP32) || defined(__AVR_ATmega2560__) || defined(__AVR_ATmega1280__)
+  if (queueLength >= QueueSize) {
+    Log.warning(F("%d Doc(s) in queue, doc blocked" CR), queueLength);
+    blockedMessages++;
     return;
   }
   Log.trace(F("Enqueue JSON" CR));
-#if defined(ESP8266) || defined(ESP32) || defined(__AVR_ATmega2560__) || defined(__AVR_ATmega1280__)
   JsonBundle bundle;
   bundle.doc = jsonDoc;
   jsonQueue.push(bundle);
@@ -484,21 +492,21 @@ void buildTopicFromId(JsonObject& Jsondata, const char* origin) {
 
 // Empty the documents queue
 void emptyQueue() {
-  while (true) {
+  queueLength = jsonQueue.size();
-    JsonBundle bundle;
+  if (queueLength > maxQueueLength) {
-
+    maxQueueLength = queueLength;
-    if (jsonQueue.empty()) {
-      break;
   }
+  if (queueLength == 0) {
+    return;
+  }
+  JsonBundle bundle;
   Log.trace(F("Dequeue JSON" CR));
   bundle = jsonQueue.front();
   jsonQueue.pop();
-
   JsonObject obj = bundle.doc.as<JsonObject>();
   pubMainCore(obj);
   queueLengthSum++;
 }
-}
 #else
 void emptyQueue() {}
 #endif
@@ -523,6 +531,7 @@ void pub(const char* topicori, const char* payload, bool retainFlag) {
  */
 void pub(const char* topicori, JsonObject& data) {
   String dataAsString = "";
+  bool ret = sensor_Retain;
 
 #if defined(ESP8266) || defined(ESP32)
 #  if message_UTCtimestamp == true
@@ -532,6 +541,10 @@ void pub(const char* topicori, JsonObject& data) {
   data["unixtime"] = unixtimestamp();
 #  endif
 #endif
+  if (data.containsKey("retain") && data["retain"].is<bool>()) {
+    ret = data["retain"];
+    data.remove("retain");
+  }
   if (data.containsKey("origin")) { // temporary, in the future use this instead of topicori
     data.remove("origin");
   }
@@ -559,7 +572,7 @@ void pub(const char* topicori, JsonObject& data) {
 
 #if jsonPublishing
   Log.trace(F("jsonPubl - ON" CR));
-  pubMQTT(topic, dataAsString.c_str());
+  pubMQTT(topic.c_str(), dataAsString.c_str(), ret);
   pubWebUI(topicori, data);
 #endif
 
@@ -2026,14 +2039,20 @@ void loop() {
 #  endif
       }
       if (now > (timer_sys_checks + (TimeBetweenCheckingSYS * 1000)) || !timer_sys_checks) {
-#  if defined(ESP32) && defined(MQTT_HTTPS_FW_UPDATE)
-        checkForUpdates();
-#  endif
 #  if defined(ESP8266) || defined(ESP32)
 #    if message_UTCtimestamp || message_unixtimestamp
         syncNTP();
 #    endif
 #  endif
+        if (!timer_sys_checks) { // Update check at start up only
+#  if defined(ESP32) && defined(MQTT_HTTPS_FW_UPDATE)
+          checkForUpdates();
+#  endif
+#  ifdef ZgatewayBT
+          ProcessLock = false; // Release BLE processes at start
+#  endif
+        }
+
         timer_sys_checks = millis();
       }
 #endif
@@ -2293,7 +2312,7 @@ void eraseAndRestart() {
 
 #if defined(ESP8266) || defined(ESP32) || defined(__AVR_ATmega2560__) || defined(__AVR_ATmega1280__)
 String stateMeasures() {
-  StaticJsonDocument<JSON_MSG_BUFFER> SYSdata;
+  StaticJsonDocument<1024> SYSdata;
 
   SYSdata["uptime"] = uptime();
 
@@ -2312,6 +2331,8 @@ String stateMeasures() {
   SYSdata["mqttsecure"] = mqtt_secure;
   SYSdata["msgproc"] = queueLengthSum;
   SYSdata["msgspeed"] = round2((float)queueLengthSum / uptime());
+  SYSdata["msgblock"] = blockedMessages;
+  SYSdata["maxqueue"] = maxQueueLength;
 #    ifdef ESP32
   minFreeMem = ESP.getMinFreeHeap();
   SYSdata["minfreemem"] = minFreeMem;
@@ -2593,9 +2614,6 @@ String latestVersion;
  */
 bool checkForUpdates() {
   Log.notice(F("Update check, free heap: %d"), ESP.getFreeHeap());
-#      ifdef ZGatewayBT
-  ProcessLock = true;
-#      endif
   HTTPClient http;
   http.setTimeout((GeneralTimeOut - 1) * 1000); // -1 to avoid WDT
   http.setFollowRedirects(HTTPC_STRICT_FOLLOW_REDIRECTS);
@@ -2616,6 +2634,7 @@ bool checkForUpdates() {
     Log.error(F("Error on HTTP request"));
   }
   http.end(); //Free the resources
+  Log.notice(F("Update check done, free heap: %d"), ESP.getFreeHeap());
   if (jsondata.containsKey("latest_version")) {
     jsondata["installed_version"] = OMG_VERSION;
     jsondata["entity_picture"] = ENTITY_PICTURE;
@@ -2623,6 +2642,7 @@ bool checkForUpdates() {
       jsondata["release_summary"] = "";
     latestVersion = jsondata["latest_version"].as<String>();
     jsondata["origin"] = subjectRLStoMQTT;
+    jsondata["retain"] = true;
     enqueueJsonObject(jsondata);
     Log.trace(F("Update file found on server" CR));
     return true;
@@ -2630,10 +2650,6 @@ bool checkForUpdates() {
     Log.trace(F("No update file found on server" CR));
     return false;
   }
-#      ifdef ZGatewayBT
-  ProcessLock = false;
-#      endif
-  Log.notice(F("Update check done, free heap: %d"), ESP.getFreeHeap());
 }
 
 #    else