fix(mqtt): publish QoS0 synchronously to bypass ~1 msg/s outbox drain
The esp-mqtt task drains only one QUEUED outbox item per loop iteration, and each iteration blocks up to MQTT_POLL_READ_TIMEOUT_MS (1s) on esp_transport_poll_read. With little inbound traffic that caps throughput at ~1 message/second per connection, so even a light packet rate (~1.2/s) outruns the drain: the outbox pins at its cap and ~20-30% of QoS0 packets are dropped as backpressure. The poll timeout is a compile-time constant baked into the precompiled esp-mqtt lib, so the async drain rate cannot be raised on the Arduino/IDF 4.4 toolchain. Route QoS0 packet publishes through esp_mqtt_client_publish() (async=false) so they write straight to the socket, bypassing the outbox drain entirely — QoS0 no longer touches the outbox. QoS1 status keeps the async/outbox + retransmit path. The esp-mqtt task releases its API lock before the poll, so a synchronous publish from the (Core-0, prio-1) MQTT task acquires the lock and writes immediately; a stalled socket blocks only that task (mesh RX on Core 1 and the WiFi/TCP stack are unaffected), bounded by a new setNetworkTimeout() lowered to 2500ms so a first stall fails fast and flips the slot to disconnected. The outbox cap from the previous commit stays as a dormant safety net. Retools the MQTT_DEBUG diagnostic from outbox size/drops (now always ~0) to per-slot publish ok/err counts, the live signal for delivery health, with 1-based slot numbering to match the status line.
This commit is contained in:
1 parent
b7c145929f
commit
53c39dc282
3 files changed
+103
-39
No files matched your search
@@ -48,6 +48,17 @@ PsychicMqttClient &PsychicMqttClient::setKeepAlive(int keepAlive)
|
||||
return *this;
|
||||
}
|
||||
|
||||
PsychicMqttClient &PsychicMqttClient::setNetworkTimeout(int timeoutMs)
|
||||
{
|
||||
#if ESP_IDF_VERSION_MAJOR == 5
|
||||
_mqtt_cfg.network.timeout_ms = timeoutMs;
|
||||
#else
|
||||
_mqtt_cfg.network_timeout_ms = timeoutMs;
|
||||
#endif
|
||||
_config_dirty = true;
|
||||
return *this;
|
||||
}
|
||||
|
||||
PsychicMqttClient &PsychicMqttClient::setAutoReconnect(bool reconnect)
|
||||
{
|
||||
#if ESP_IDF_VERSION_MAJOR == 5
|
||||
@@ -554,12 +565,19 @@ int PsychicMqttClient::publish(const char *topic, int qos, bool retain, const ch
|
||||
|
||||
ESP_LOGV(TAG, "Enqueuing message to topic %s with QoS %d", topic, qos);
|
||||
bool store_in_outbox = true;
|
||||
return esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, store_in_outbox);
|
||||
int result = esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, store_in_outbox);
|
||||
if (result < 0) _publish_err++; else _publish_ok++;
|
||||
return result;
|
||||
}
|
||||
else
|
||||
{
|
||||
ESP_LOGV(TAG, "Publishing message to topic %s with QoS %d", topic, qos);
|
||||
return esp_mqtt_client_publish(_client, topic, payload, length, qos, retain);
|
||||
// Synchronous write (used for QoS0 packet publishes). A negative result is a real
|
||||
// send failure (socket error / network_timeout on a stalled link), tracked here so
|
||||
// callers can surface delivery health without per-message logging.
|
||||
int result = esp_mqtt_client_publish(_client, topic, payload, length, qos, retain);
|
||||
if (result < 0) _publish_err++; else _publish_ok++;
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -601,6 +619,16 @@ unsigned long PsychicMqttClient::getOutboxDrops()
|
||||
return _outbox_drops;
|
||||
}
|
||||
|
||||
unsigned long PsychicMqttClient::getPublishOk()
|
||||
{
|
||||
return _publish_ok;
|
||||
}
|
||||
|
||||
unsigned long PsychicMqttClient::getPublishErr()
|
||||
{
|
||||
return _publish_err;
|
||||
}
|
||||
|
||||
void PsychicMqttClient::_onMqttEventStatic(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data)
|
||||
{
|
||||
// Since this is a static function, we need to cast the first argument (void*) back to the class instance type
|
||||
|
||||
@@ -126,6 +126,18 @@ public:
|
||||
*/
|
||||
PsychicMqttClient &setAutoReconnect(bool reconnect = true);
|
||||
|
||||
/**
|
||||
* @brief Sets the network operation timeout in milliseconds. esp-mqtt aborts a
|
||||
* network read/write (including a synchronous publish's socket write) if it does
|
||||
* not complete within this window. A lower value bounds how long a synchronous
|
||||
* QoS0 publish can block on a stalled/half-open socket before failing and letting
|
||||
* the slot flip to disconnected. esp-mqtt's default is 10000.
|
||||
*
|
||||
* @param timeoutMs Network timeout in milliseconds.
|
||||
* @return A reference to the PsychicMqttClient instance.
|
||||
*/
|
||||
PsychicMqttClient &setNetworkTimeout(int timeoutMs);
|
||||
|
||||
/**
|
||||
* @brief Sets the retransmit timeout for unacknowledged QoS 1/2 messages.
|
||||
* esp-mqtt resends an unacked PUBLISH (DUP flag set) every time this
|
||||
@@ -446,6 +458,19 @@ public:
|
||||
*/
|
||||
unsigned long getOutboxDrops();
|
||||
|
||||
/**
|
||||
* @brief Cumulative count of publishes that were accepted (enqueued or written
|
||||
* successfully). Monotonic. Pair with getPublishErr() for delivery-health stats.
|
||||
*/
|
||||
unsigned long getPublishOk();
|
||||
|
||||
/**
|
||||
* @brief Cumulative count of publishes that failed (negative return from the
|
||||
* synchronous write or async enqueue — socket error / network timeout). Monotonic.
|
||||
* A rising value indicates the uplink is dropping publishes.
|
||||
*/
|
||||
unsigned long getPublishErr();
|
||||
|
||||
private:
|
||||
esp_mqtt_client_handle_t _client = nullptr;
|
||||
esp_mqtt_client_config_t _mqtt_cfg;
|
||||
@@ -459,6 +484,9 @@ private:
|
||||
size_t _outbox_limit = 0;
|
||||
// Cumulative count of QoS0 publishes dropped because the outbox hit the cap.
|
||||
unsigned long _outbox_drops = 0;
|
||||
// Cumulative publish accept/fail counts (any QoS, sync or async path).
|
||||
unsigned long _publish_ok = 0;
|
||||
unsigned long _publish_err = 0;
|
||||
|
||||
// Multipart message reassembly. _buffer is lazily allocated at connect() time
|
||||
// to match the configured buffer size, then reused for the client's lifetime.
|
||||
|
||||
@@ -1752,17 +1752,23 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload
|
||||
return false;
|
||||
}
|
||||
|
||||
// QoS 0 for the high-rate packet/raw publish paths: no PUBACK, so delivery is
|
||||
// best-effort. These still enter the esp-mqtt outbox (store=true, needed so packet
|
||||
// topics keep flowing), but the client bounds that outbox via setOutboxLimit() to
|
||||
// stop QoS0 frames accumulating on internal heap during a stalled uplink — over the
|
||||
// cap, publish() returns -2 and the queue retry/drop path below handles it. QoS 1 is
|
||||
// used only for low-rate retained status messages where delivery matters.
|
||||
// Publish path by QoS:
|
||||
// - QoS 0 (high-rate packets/raw): SYNCHRONOUS (async=false → esp_mqtt_client_publish),
|
||||
// which writes straight to the socket. The async/outbox path drains only one queued
|
||||
// item per esp-mqtt task loop (~1 msg/s/conn, gated by the 1s poll_read), so under
|
||||
// even light packet load the outbox pins at its cap and drops ~20-30%. A synchronous
|
||||
// write bypasses that drain ceiling entirely and does not store in the outbox. It can
|
||||
// block the (Core-0, prio-1) MQTT task on a stalled socket, but only up to
|
||||
// network_timeout_ms (lowered in optimizeMqttClientConfig); mesh RX (Core 1) and the
|
||||
// WiFi/TCP stack (higher-prio system tasks) are unaffected, and a failed write flips
|
||||
// the slot to disconnected so subsequent packets skip it.
|
||||
// - QoS 1 (low-rate retained status): async, so it keeps the durable outbox + retransmit.
|
||||
//
|
||||
// esp_mqtt_client_enqueue return convention: QoS 0 returns msg_id == 0 on success
|
||||
// (no tracking since there's no PUBACK); QoS 1/2 return a positive msg_id. Negative
|
||||
// values (-1 generic failure, -2 outbox full / over cap) are the only actual failures.
|
||||
int result = slot.client->publish(topic, qos, retained, payload, strlen(payload), true);
|
||||
// Return convention: QoS 0 sync publish returns msg_id == 0 on success (no PUBACK
|
||||
// tracking). Negative values (-1 write/failure) are the only actual failures; the queue
|
||||
// retry/drop path below handles them.
|
||||
bool async = (qos > 0);
|
||||
int result = slot.client->publish(topic, qos, retained, payload, strlen(payload), async);
|
||||
if (result < 0) {
|
||||
// QoS0 packet/raw publishes are best-effort and may be retried from the
|
||||
// bridge queue; avoid logging transient first-attempt failures here.
|
||||
@@ -3421,16 +3427,17 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_
|
||||
|
||||
client->setBufferSize(MQTT_CLIENT_BUFFER_SIZE);
|
||||
|
||||
// Bound the esp-mqtt outbox for high-rate QoS0 packet/raw publishes. Those go in
|
||||
// with store=true (so packet topics keep flowing) but the outbox has no size limit
|
||||
// of its own — it frees entries only on send-ack or ~30s expiry, so a stalled uplink
|
||||
// lets QoS0 frames accumulate on internal heap without bound. Above this cap the
|
||||
// client drops the new QoS0 message (returns -2), and processPacketQueue's existing
|
||||
// retry/drop path applies backpressure. Non-PSRAM is the fragmentation-sensitive case
|
||||
// (outbox lives on internal heap), so it gets the tighter cap. Under healthy operation
|
||||
// the outbox drains to ~0 in ms, so this only bites during a stall. Note: esp-mqtt's
|
||||
// own outbox.limit config is not used — its enqueue path does not reliably enforce it
|
||||
// for QoS0, and this app-level guard fires before enqueue regardless of IDF version.
|
||||
// Bound how long a synchronous QoS0 publish (see publishToSlot) can block the MQTT
|
||||
// task on a stalled/half-open socket before esp-mqtt aborts the write. Default is 10s;
|
||||
// 2.5s lets a first stall resolve fast (write fails → slot flips to disconnected →
|
||||
// subsequent packets skip it) without holding up publishing to the other slots. Mesh
|
||||
// RX (Core 1) and the WiFi/TCP stack are unaffected by this block regardless.
|
||||
client->setNetworkTimeout(2500);
|
||||
|
||||
// Dormant safety net: cap the esp-mqtt outbox for any residual async QoS0 path. QoS0
|
||||
// packets now publish synchronously (store=false, no outbox), so this normally never
|
||||
// engages, but it bounds internal-heap growth if a QoS0 message ever takes the async
|
||||
// path. Non-PSRAM (outbox on internal heap) gets the tighter cap.
|
||||
#if defined(BOARD_HAS_PSRAM)
|
||||
client->setOutboxLimit(16384);
|
||||
#else
|
||||
@@ -3454,30 +3461,31 @@ void MQTTBridge::optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_
|
||||
}
|
||||
|
||||
void MQTTBridge::logMemoryStatus() {
|
||||
// The esp-mqtt outbox is the structure that grew unbounded before the cap. The cap
|
||||
// (setOutboxLimit) is enforced PER CLIENT, so log it per slot — a summed total hides
|
||||
// whether any single slot has blown past its cap. Each entry is size/limit; drops is
|
||||
// the cumulative count of QoS0 messages rejected at the cap (backpressure firing).
|
||||
// Under a healthy uplink size sits near 0; under a stall/saturation it should plateau
|
||||
// at limit (not climb) and drops should increase.
|
||||
char outbox_detail[160];
|
||||
// QoS0 packets now publish synchronously, so the outbox stays ~0 and is only a sanity
|
||||
// check (a non-zero total would mean the QoS1 status path is backing up or the dormant
|
||||
// async cap engaged). The live signal is per-slot publish health: ok = cumulative
|
||||
// accepted writes, err = cumulative failures (socket error / network_timeout on a
|
||||
// stalled link). A rising err on a slot means that broker's uplink is dropping packets;
|
||||
// ok climbing with err flat is healthy delivery.
|
||||
char pub_detail[200];
|
||||
size_t pos = 0;
|
||||
size_t outbox_total = 0;
|
||||
outbox_detail[0] = '\0';
|
||||
pub_detail[0] = '\0';
|
||||
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
|
||||
if (_slots[i].client) {
|
||||
size_t sz = _slots[i].client->getOutboxSize();
|
||||
outbox_total += sz;
|
||||
pos += snprintf(outbox_detail + pos, sizeof(outbox_detail) - pos, "%ss%d=%u/%u(d%lu)",
|
||||
pos ? " " : "", i, (unsigned)sz,
|
||||
(unsigned)_slots[i].client->getOutboxLimit(),
|
||||
_slots[i].client->getOutboxDrops());
|
||||
if (pos >= sizeof(outbox_detail)) break;
|
||||
outbox_total += _slots[i].client->getOutboxSize();
|
||||
if (_slots[i].enabled) {
|
||||
pos += snprintf(pub_detail + pos, sizeof(pub_detail) - pos, "%ss%d=%lu/%lu",
|
||||
pos ? " " : "", i + 1,
|
||||
_slots[i].client->getPublishOk(),
|
||||
_slots[i].client->getPublishErr());
|
||||
if (pos >= sizeof(pub_detail)) break;
|
||||
}
|
||||
}
|
||||
}
|
||||
MQTT_DEBUG_PRINTLN("Memory: Free=%d, Max=%d, Queue=%d/%d, Outbox=%u [%s]",
|
||||
MQTT_DEBUG_PRINTLN("Memory: Free=%d, Max=%d, Queue=%d/%d, Outbox=%u | pub(ok/err) %s",
|
||||
ESP.getFreeHeap(), ESP.getMaxAllocHeap(), _queue_count, MAX_QUEUE_SIZE,
|
||||
(unsigned)outbox_total, outbox_detail);
|
||||
(unsigned)outbox_total, pub_detail);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
Reference in new issue
Block a user