I'm trying to get a automatic OTA system running where the Particle automatically updates slave devices connected to it by Streaming the Assets that are included with the firmware.
However I've run into an issue when my code has hit a memory boundary causing the application thread to hang. It seems that looking at the Firmware core code that the Inflator stream uses a 32k window but is using the Free Heap vs the Largest Free Block to determin if it can do it or not using heap.
Below is a working example extract of my code that simulates the issue. It fragments the memory and then tries to read the asset. Instead of either streaming the inflate or allocating smaller memory or just outright erroring it hangs the thread.
/*
* Particle compressed asset read / heap fragmentation repro.
*
* project.properties:
* assetOtaDir=assets
*
* assets:
mkdir -p assets
python3 - <<'PY'
from pathlib import Path
Path("assets").mkdir(exist_ok=True)
# Compressible dummy payloads, similar sizes to the real OTA assets.
Path("assets/firmware_350k.bin").write_bytes((b"FIRMWARE0123456789" * 20480)[:350 * 1024])
Path("assets/firmware_20k.bin").write_bytes((b"SMALL0123456789" * 2048)[:20 * 1024])
PY
*
* Build/flash for a device with asset support. Open Serial at 115200.
*/
#include "Particle.h"
#include "core_hal.h"
SYSTEM_MODE(SEMI_AUTOMATIC);
SYSTEM_THREAD(ENABLED);
namespace {
constexpr bool CONNECT_CLOUD = true;
constexpr uint32_t SERIAL_WAIT_MS = 10000;
constexpr uint32_t READ_BUFFER_SIZE = 384;
constexpr uint32_t WDT_KICK_INTERVAL_MS = 10000;
constexpr bool EXTERNAL_WDT_ENABLED = true;
constexpr pin_t EXTERNAL_WDT_KICK_PIN = D22;
constexpr uint32_t EXTERNAL_WDT_KICK_INTERVAL_MS = 250;
constexpr const char* LARGE_ASSET = "firmware_350k.bin";
constexpr const char* SMALL_ASSET = "firmware_20k.bin";
constexpr uint32_t HEAP_PRESSURE_TARGET_FREE = 54000;
constexpr uint32_t HEAP_PRESSURE_MIN_LARGEST_BLOCK = 42000;
constexpr size_t MAX_HEAP_BLOCKS = 96;
void* heapBlocks[MAX_HEAP_BLOCKS] = {};
bool externalWdtState = false;
uint32_t externalWdtLastKick = 0;
void kickExternalWatchdogIfDue(bool force = false) {
if (!EXTERNAL_WDT_ENABLED) {
return;
}
const uint32_t now = millis();
if (!force && now - externalWdtLastKick < EXTERNAL_WDT_KICK_INTERVAL_MS) {
return;
}
externalWdtLastKick = now;
externalWdtState = !externalWdtState;
digitalWrite(EXTERNAL_WDT_KICK_PIN, externalWdtState ? HIGH : LOW);
}
void setupExternalWatchdog() {
if (!EXTERNAL_WDT_ENABLED) {
return;
}
pinMode(EXTERNAL_WDT_KICK_PIN, OUTPUT);
digitalWrite(EXTERNAL_WDT_KICK_PIN, LOW);
externalWdtState = false;
externalWdtLastKick = 0;
kickExternalWatchdogIfDue(true);
}
bool waitForSerialWithWatchdog(uint32_t timeoutMs) {
const uint32_t deadline = millis() + timeoutMs;
while (!Serial.isConnected() && (int32_t)(millis() - deadline) < 0) {
kickExternalWatchdogIfDue();
delay(10);
}
return Serial.isConnected();
}
bool waitForCloudWithWatchdog(uint32_t timeoutMs) {
const uint32_t deadline = millis() + timeoutMs;
while (!Particle.connected() && (int32_t)(millis() - deadline) < 0) {
kickExternalWatchdogIfDue();
Particle.process();
delay(10);
}
return Particle.connected();
}
void logMemory(const char* label) {
runtime_info_t info = {};
info.size = sizeof(info);
HAL_Core_Runtime_Info(&info, nullptr);
Serial.printf("[%10lu][%lu][%lu][%lu] %s\r\n",
millis(),
(unsigned long)info.freeheap,
(unsigned long)info.largest_free_block_heap,
(unsigned long)info.max_used_heap,
label);
}
runtime_info_t getRuntimeInfo() {
runtime_info_t info = {};
info.size = sizeof(info);
HAL_Core_Runtime_Info(&info, nullptr);
return info;
}
void fragmentHeap() {
logMemory("heap pressure: before");
size_t allocated = 0;
for (size_t i = 0; i < MAX_HEAP_BLOCKS; ++i) {
kickExternalWatchdogIfDue();
runtime_info_t info = getRuntimeInfo();
if (info.freeheap <= HEAP_PRESSURE_TARGET_FREE ||
info.largest_free_block_heap <= HEAP_PRESSURE_MIN_LARGEST_BLOCK) {
break;
}
uint32_t excess = info.freeheap - HEAP_PRESSURE_TARGET_FREE;
size_t size = excess > 8192 ? 4096 : excess > 2048 ? 1024 : 256;
uint8_t* block = new(std::nothrow) uint8_t[size];
if (!block && size > 256) {
size = 256;
block = new(std::nothrow) uint8_t[size];
}
heapBlocks[i] = block;
if (block) {
memset(block, (int)i, size);
allocated++;
} else {
break;
}
}
// Add a little fragmentation without destroying the large tail block. The
// target is similar to the real failure: low free heap, but not immediate
// SYSTEM_ERROR_NO_MEMORY from seek().
for (size_t i = 2; i < allocated; i += 8) {
kickExternalWatchdogIfDue();
delete[] static_cast<uint8_t*>(heapBlocks[i]);
heapBlocks[i] = nullptr;
}
logMemory("heap pressure: after");
}
void releaseHeapFragments() {
for (size_t i = 0; i < MAX_HEAP_BLOCKS; ++i) {
kickExternalWatchdogIfDue();
delete[] static_cast<uint8_t*>(heapBlocks[i]);
heapBlocks[i] = nullptr;
}
logMemory("heap pressure: released");
}
void listAssets() {
logMemory("assets: before System.assetsAvailable");
spark::Vector<ApplicationAsset> assets = System.assetsAvailable();
Serial.printf("available=%u\r\n", (unsigned)assets.size());
for (auto& asset : assets) {
Serial.printf("asset name=%s size=%lu storage=%lu\r\n",
asset.name().c_str(),
(unsigned long)asset.size(),
(unsigned long)asset.storageSize());
}
logMemory("assets: after System.assetsAvailable");
}
bool readAssetToNullBuffer(const char* name) {
Serial.printf("\r\n--- read %s ---\r\n", name);
logMemory("read: start");
spark::Vector<ApplicationAsset> assets = System.assetsAvailable();
ApplicationAsset asset;
for (auto& candidate : assets) {
if (candidate.name().equals(name)) {
asset = candidate;
break;
}
}
if (!asset.isValid()) {
Serial.printf("read: asset not found: %s\r\n", name);
return false;
}
Serial.printf("read: found name=%s size=%lu storage=%lu\r\n",
asset.name().c_str(),
(unsigned long)asset.size(),
(unsigned long)asset.storageSize());
logMemory("read: before seek");
int seekResult = asset.seek(0);
Serial.printf("read: seek result=%d\r\n", seekResult);
logMemory("read: after seek");
if (seekResult < 0) {
return false;
}
uint8_t buffer[READ_BUFFER_SIZE];
uint32_t bytesRead = 0;
uint32_t lastKick = millis();
while (true) {
kickExternalWatchdogIfDue();
int n = asset.read(reinterpret_cast<char*>(buffer), sizeof(buffer));
if (n == SYSTEM_ERROR_END_OF_STREAM) {
break;
}
if (n < 0) {
Serial.printf("read: error=%d bytes=%lu\r\n", n, (unsigned long)bytesRead);
return false;
}
if (n == 0) {
Serial.printf("read: zero-byte read bytes=%lu\r\n", (unsigned long)bytesRead);
break;
}
bytesRead += (uint32_t)n;
if ((bytesRead % (32 * 1024)) < READ_BUFFER_SIZE) {
Serial.printf("read: progress bytes=%lu\r\n", (unsigned long)bytesRead);
logMemory("read: progress");
}
if (millis() - lastKick >= WDT_KICK_INTERVAL_MS) {
Particle.process();
lastKick = millis();
}
}
logMemory("read: after eof");
Serial.printf("read: complete bytes=%lu\r\n", (unsigned long)bytesRead);
asset.reset();
logMemory("read: after reset");
return true;
}
} // namespace
void setup() {
setupExternalWatchdog();
Serial.begin(115200);
waitForSerialWithWatchdog(SERIAL_WAIT_MS);
delay(500);
Serial.println();
Serial.println("Particle compressed asset heap repro");
Serial.println("[millis][freeheap][largest_free_block][max_used_heap] message");
Serial.printf("external watchdog: %s pin=%d interval=%lums\r\n",
EXTERNAL_WDT_ENABLED ? "enabled" : "disabled",
(int)EXTERNAL_WDT_KICK_PIN,
(unsigned long)EXTERNAL_WDT_KICK_INTERVAL_MS);
logMemory("setup: boot");
if (CONNECT_CLOUD) {
Serial.println("cloud: connecting");
Particle.connect();
if (waitForCloudWithWatchdog(120000)) {
logMemory("cloud: connected");
} else {
logMemory("cloud: connection timeout");
}
}
listAssets();
fragmentHeap();
readAssetToNullBuffer(LARGE_ASSET);
readAssetToNullBuffer(SMALL_ASSET);
readAssetToNullBuffer(LARGE_ASSET);
releaseHeapFragments();
Serial.println("setup: done");
}
void loop() {
kickExternalWatchdogIfDue();
static uint32_t last = 0;
if (millis() - last >= 5000) {
last = millis();
logMemory("loop: idle");
}
}