refactor(mqtt): improve command handling and memory pressure management
Enhance the command handling in CommonCLI_Observer to clarify the compilation behavior under different configurations. Additionally, optimize memory pressure checks in MQTTBridge to reduce unnecessary heap walks, improving performance during sustained memory pressure. This change ensures that publishes are skipped only when necessary, enhancing overall stability.
This commit is contained in:
1 parent
1b84311183
commit
2d55b95ee5
4 files changed
+37
-19
No files matched your search
@@ -3,10 +3,12 @@
|
||||
// only two small delegation hooks. These are CommonCLI member functions, so they
|
||||
// retain full access to _prefs/_callbacks/_board/savePrefs() with no re-plumbing.
|
||||
//
|
||||
// Behavior is intentionally identical to the previously-inlined branches: MQTT
|
||||
// commands keep their WITH_MQTT_BRIDGE guard; alert/SNMP commands remain unguarded.
|
||||
// Each handler returns true if it recognized the command, false to fall through to
|
||||
// the base get/set parser in CommonCLI.cpp.
|
||||
// Behavior is intentionally identical to the previously-inlined branches. NOTE:
|
||||
// the entire body of each set/get handler here is compiled under WITH_MQTT_BRIDGE
|
||||
// (see the #ifdef at the top of each), so on an observer build without the bridge
|
||||
// the WiFi/timezone/alert/SNMP commands compile out too — they are not guarded
|
||||
// independently of the MQTT commands. Each handler returns true if it recognized
|
||||
// the command, false to fall through to the base get/set parser in CommonCLI.cpp.
|
||||
|
||||
#include <Arduino.h>
|
||||
#include "CommonCLI.h"
|
||||
@@ -430,6 +432,16 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
|
||||
} else {
|
||||
sprintf(reply, "OK - slot %d preset: %s", slot + 1, preset_name);
|
||||
}
|
||||
// Slots beyond RUNTIME_MQTT_SLOTS are still stored (so they take effect
|
||||
// if the prefs move to a PSRAM board) but won't connect on this
|
||||
// hardware — say so, or the operator waits for a connection that never
|
||||
// comes (A15).
|
||||
if (slot >= RUNTIME_MQTT_SLOTS) {
|
||||
size_t used = strlen(reply);
|
||||
if (used < 158) {
|
||||
snprintf(reply + used, 160 - used, " (slot inactive on this hardware)");
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
strcpy(reply, "Error: unknown preset. Use 'get mqtt.presets'");
|
||||
|
||||
@@ -134,7 +134,7 @@ static const int MQTT_PRESET_COUNT = 28;
|
||||
|
||||
// Built-in preset definitions (stored in flash)
|
||||
static const MQTTPresetDef MQTT_PRESETS[MQTT_PRESET_COUNT] = {
|
||||
// name url server rootCA auth topic keepalive tls enabled interval user pass
|
||||
// name url audience rootCA auth topic tokenLife retain keepAlive user pass
|
||||
{ "analyzer-us", "wss://mqtt-us-v1.letsmesh.net:443/mqtt", "mqtt-us-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55, nullptr, nullptr },
|
||||
{ "analyzer-eu", "wss://mqtt-eu-v1.letsmesh.net:443/mqtt", "mqtt-eu-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55, nullptr, nullptr },
|
||||
{ "nz-analyzer", "wss://meshcore-mqtt-1.baird.io:443", "meshcore-mqtt-1.baird.io", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55, nullptr, nullptr },
|
||||
|
||||
@@ -3118,19 +3118,25 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx,
|
||||
static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 8000;
|
||||
#endif
|
||||
unsigned long now = millis();
|
||||
// Re-sample max-alloc at most once per interval and cache the verdict.
|
||||
// getMaxAllocHeap() walks the heap free-list, so it must not run per packet.
|
||||
// The previous code only advanced _last_memory_check on the healthy path, so
|
||||
// under sustained pressure the guard stayed open and it walked the heap on
|
||||
// EVERY packet — the opposite of throttling (A15). Caching the verdict keeps
|
||||
// the "skip publishes while memory is low" protection but pays for the walk
|
||||
// only once per interval; _last_memory_check is now advanced on both paths.
|
||||
if (now - _last_memory_check > 5000) {
|
||||
size_t max_alloc = ESP.getMaxAllocHeap();
|
||||
if (max_alloc < PUBLISH_SKIP_MAX_ALLOC_THRESHOLD) {
|
||||
_skipped_publishes++;
|
||||
static unsigned long last_skip_log = 0;
|
||||
if (now - last_skip_log > 60000) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT: Skipping publish due to memory pressure (Max alloc: %d, threshold: %d, skipped: %d)",
|
||||
max_alloc, (int)PUBLISH_SKIP_MAX_ALLOC_THRESHOLD, _skipped_publishes);
|
||||
last_skip_log = now;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
_last_memory_check = now;
|
||||
size_t max_alloc = ESP.getMaxAllocHeap();
|
||||
_memory_pressure = (max_alloc < PUBLISH_SKIP_MAX_ALLOC_THRESHOLD);
|
||||
if (_memory_pressure) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT: memory pressure, skipping publishes (Max alloc: %d, threshold: %d, skipped: %d)",
|
||||
(int)max_alloc, (int)PUBLISH_SKIP_MAX_ALLOC_THRESHOLD, _skipped_publishes);
|
||||
}
|
||||
}
|
||||
if (_memory_pressure) {
|
||||
_skipped_publishes++;
|
||||
return false;
|
||||
}
|
||||
#endif
|
||||
|
||||
@@ -3386,7 +3392,6 @@ void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) {
|
||||
MQTTPacketQueuePolicy::enqueueAction(
|
||||
static_cast<size_t>(_queue_count), static_cast<size_t>(MAX_QUEUE_SIZE));
|
||||
if (action == MQTTPacketQueuePolicy::EnqueueAction::EvictOldestThenEnqueue) {
|
||||
QueuedPacket& oldest = _packet_queue[_queue_head];
|
||||
MQTT_DEBUG_PRINTLN("Queue full, dropping oldest packet (queue size: %d)", _queue_count);
|
||||
dequeuePacket();
|
||||
} else if (action == MQTTPacketQueuePolicy::EnqueueAction::Reject) {
|
||||
|
||||
@@ -323,6 +323,7 @@ private:
|
||||
// the MQTT memory-defrag work — persistent MQTT clients no longer churn
|
||||
// the heap, so gray-zone / critical-restart trackers are unnecessary.
|
||||
unsigned long _last_memory_check;
|
||||
bool _memory_pressure = false; // Cached max-alloc verdict; re-sampled at most once per interval in publishPacket() so the heap walk isn't paid per-packet under pressure
|
||||
int _skipped_publishes; // Exposed via SNMP; count of publishes skipped when max_alloc is too low
|
||||
|
||||
// Status publish retry tracking
|
||||
@@ -474,7 +475,7 @@ public:
|
||||
* "set mqttN.preset <name>". Handles teardown of old connection and
|
||||
* setup of new one.
|
||||
*
|
||||
* @param slot_index Slot index (0-2)
|
||||
* @param slot_index Slot index (0 to RUNTIME_MQTT_SLOTS-1)
|
||||
* @param preset_name Preset name: "analyzer-us", "analyzer-eu", "nz-analyzer", "meshmapper", "custom", "none"
|
||||
*/
|
||||
void setSlotPreset(int slot_index, const char* preset_name);
|
||||
@@ -484,7 +485,7 @@ public:
|
||||
* Configure custom broker settings for a slot. Only applies when the
|
||||
* slot's preset is "custom".
|
||||
*
|
||||
* @param slot_index Slot index (0-2)
|
||||
* @param slot_index Slot index (0 to RUNTIME_MQTT_SLOTS-1)
|
||||
* @param host Broker hostname
|
||||
* @param port Broker port
|
||||
* @param username MQTT username (empty for anonymous)
|
||||
|
||||
Reference in new issue
Block a user