[SYS] Fixed data corruption associated with jsonQueue enqueueing and dequeuing (#2035)

* Changed jsonQueue to use json String rather than json Object
Added Mutex protection to emptyQueue()
Move Mutex protection to within enqueueJsonObject to shorten lock time
This commit is contained in:
puterboy authored and GitHub committed 2024-08-29 07:45:56 -05:00
1 parent 830d58f98e
commit 1f42075ec8
1 file changed
+44 -31
+44 -31
View File
@@ -89,7 +89,7 @@ struct JsonBundle {
StaticJsonDocument<JSON_MSG_BUFFER> doc; StaticJsonDocument<JSON_MSG_BUFFER> doc;
}; };
std::queue<JsonBundle> jsonQueue; std::queue<String> jsonQueue;
#ifdef ESP32 #ifdef ESP32
# include <driver/adc.h> # include <driver/adc.h>
@@ -485,36 +485,46 @@ void jsonDispatch(JsonObject& data) {
} }
// Add a document to the queue // Add a document to the queue
void enqueueJsonObject(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc) { boolean enqueueJsonObject(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc, int timeout) {
if (jsonDoc.size() == 0) { if (jsonDoc.size() == 0) {
Log.error(F("Empty JSON, skipping" CR)); Log.error(F("Empty JSON, skipping" CR));
return; return true;
} }
if (queueLength >= QueueSize) { if (queueLength >= QueueSize) {
Log.warning(F("%d Doc(s) in queue, doc blocked" CR), queueLength); Log.warning(F("%d Doc(s) in queue, doc blocked" CR), queueLength);
blockedMessages++; blockedMessages++;
return; return false;
} }
JsonBundle bundle; Log.trace(F("Enqueue JSON" CR));
bundle.doc = jsonDoc; String jsonString;
jsonQueue.push(bundle); serializeJson(jsonDoc, jsonString);
Log.trace(F("Queue length: %d" CR), jsonQueue.size());
}
#ifdef ESP32 #ifdef ESP32
// Semaphore check before enqueueing a document // Semaphore check before enqueueing a document
bool handleJsonEnqueue(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc, int timeout) { if (xSemaphoreTake(xQueueMutex, pdMS_TO_TICKS(timeout)) == pdFALSE) {
if (xSemaphoreTake(xQueueMutex, pdMS_TO_TICKS(timeout))) {
enqueueJsonObject(jsonDoc);
xSemaphoreGive(xQueueMutex);
return true;
} else {
Log.error(F("xQueueMutex not taken" CR)); Log.error(F("xQueueMutex not taken" CR));
blockedMessages++; blockedMessages++;
return false; return false;
} }
#endif
jsonQueue.push(jsonString);
#ifdef ESP32
xSemaphoreGive(xQueueMutex);
#endif
Log.trace(F("Queue length: %d" CR), jsonQueue.size());
return true;
} }
// Semaphore check before enqueueing a document
bool handleJsonEnqueue(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc, int timeout) {
return enqueueJsonObject(jsonDoc, timeout);
}
// Semaphore check before enqueueing a document with default timeout QueueSemaphoreTimeOutLoop
bool handleJsonEnqueue(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc) {
return handleJsonEnqueue(jsonDoc, QueueSemaphoreTimeOutLoop);
}
#ifdef ESP32
# include "mbedtls/sha256.h" # include "mbedtls/sha256.h"
std::string generateHash(const std::string& input) { std::string generateHash(const std::string& input) {
@@ -529,21 +539,11 @@ std::string generateHash(const std::string& input) {
return std::string(hashString); return std::string(hashString);
} }
#else #else
bool handleJsonEnqueue(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc, int timeout) {
enqueueJsonObject(jsonDoc);
return true;
}
std::string generateHash(const std::string& input) { std::string generateHash(const std::string& input) {
return "Not implemented for ESP8266"; return "Not implemented for ESP8266";
} }
#endif #endif
// Semaphore check before enqueueing a document with default timeout QueueSemaphoreTimeOutLoop
bool handleJsonEnqueue(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc) {
return handleJsonEnqueue(jsonDoc, QueueSemaphoreTimeOutLoop);
}
/* /*
* Add the jsonObject id as a topic to the jsonObject origin * Add the jsonObject id as a topic to the jsonObject origin
* *
@@ -591,12 +591,25 @@ void emptyQueue() {
if (queueLength == 0) { if (queueLength == 0) {
return; return;
} }
JsonBundle bundle;
Log.trace(F("Dequeue JSON" CR)); Log.trace(F("Dequeue JSON" CR));
bundle = jsonQueue.front(); DynamicJsonDocument jsonBuffer(JSON_MSG_BUFFER);
JsonObject obj = jsonBuffer.to<JsonObject>();
#ifdef ESP32
if (xSemaphoreTake(xQueueMutex, pdMS_TO_TICKS(QueueSemaphoreTimeOutTask)) == pdFALSE) {
Log.error(F("xQueueMutex not taken" CR));
return;
}
#endif
auto error = deserializeJson(jsonBuffer, jsonQueue.front());
jsonQueue.pop(); jsonQueue.pop();
JsonObject obj = bundle.doc.as<JsonObject>(); #ifdef ESP32
jsonDispatch(obj); xSemaphoreGive(xQueueMutex);
#endif
if (error) {
Log.error(F("deserialize jsonQueue.front() failed: %s, buffer capacity: %u" CR), error.c_str(), jsonBuffer.capacity());
} else {
jsonDispatch(obj);
}
queueLengthSum++; queueLengthSum++;
} }