Reduce ESP contact save backend work with bounded buffering

This commit is contained in:
mikecarper committed 2026-10-04 08:20:28 -07:00
1 parent 85ff9e321e
commit 62d76b7bab
4 files changed
+145 -11

No files matched your search

+12 -3
View File
@@ -29,6 +29,13 @@ private:
size_t _verify_remaining = 0;
PresenceProbe _presence;
#if defined(ESP32_PLATFORM)
// Pinned ESP SPIFFS uses 256-byte pages with a 5-byte data-page header.
// This bounded stdio buffer coalesces records without the SDK's 4 KiB burst;
// one logical record can flush at most one full buffer per write pass.
static constexpr size_t ESP_WRITE_BUFFER_SIZE = 251;
static constexpr size_t ESP_VERIFY_BUFFER_SIZE = 128;
static_assert(storage::CONTACT_RECORD_SIZE <= ESP_WRITE_BUFFER_SIZE,
"Contact records must fit one bounded writer buffer");
enum class BeginStage : uint8_t {
Target, Backup, Recover, Temp, Remove, Open, Configure, Ready, Failed
};
@@ -135,7 +142,7 @@ public:
case BeginStage::Configure:
// Arduino otherwise batches small writes into a 4 KiB stdio buffer.
// Configure the newly opened stream before its first I/O.
ok = _file.setBufferSize(storage::CONTACT_RECORD_SIZE);
ok = _file.setBufferSize(ESP_WRITE_BUFFER_SIZE);
if (ok) {
_ok = true;
_begin_stage = BeginStage::Ready;
@@ -186,8 +193,10 @@ public:
ok = static_cast<bool>(_verify);
#if defined(ESP32_PLATFORM)
// fread() otherwise reads ahead up to 4 KiB while the caller requests
// only one 64-byte CRC chunk. A failed configuration cannot publish.
ok = ok && _verify.setBufferSize(64);
// only one 64-byte CRC chunk. A bounded two-chunk read-ahead buffer
// halves backend reads without growing that scratch buffer or allowing
// more than one backend refill per pass. Configuration must succeed.
ok = ok && _verify.setBufferSize(ESP_VERIFY_BUFFER_SIZE);
#endif
ok = ok && _verify.size() == _size;
if (ok) {
+1
View File
@@ -176,6 +176,7 @@ size_t File::write(const uint8_t* bytes, size_t length) {
const size_t part = std::min(count - consumed, _stdio_size - _stdio_buffer.size());
_stdio_buffer.insert(_stdio_buffer.end(), bytes + consumed, bytes + consumed + part);
consumed += part;
if (_stdio_buffer.size() == _stdio_size) flushWriteBuffer();
}
_position += count;
return count;
+80 -2
View File
@@ -48,6 +48,11 @@ using DataStoreHost = Host;
// all save/load/dirty/transaction bodies below come from production sources.
class DataStore {
FakeFilesystem* _fs;
// Advert queue behavior is covered by its dedicated production fixture;
// these hardware seams keep this test focused on contacts and disk delete.
struct { bool synchronous_io = false; } _advert_write;
void invalidateAdvertWrite(const uint8_t[], int) {}
void retireAdvertWrite(const uint8_t[], int) {}
bool _channel_load_incomplete = false;
bool _uncached_contact_load_incomplete = false;
#include "contact_write_state_under_test.h"
@@ -166,8 +171,8 @@ void boundedFullTableAndFirstSave() {
const auto passes = drain(store, h, keepContact);
assert(passes > expected.size() / 64);
assert(SPIFFS.largest_read <= 64);
assert(SPIFFS.largest_backend_read <= 64);
assert(SPIFFS.largest_backend_write <= mesh::storage::CONTACT_RECORD_SIZE);
assert(SPIFFS.largest_backend_read <= 128);
assert(SPIFFS.largest_backend_write <= 251);
assert(SPIFFS.largest_write == mesh::storage::CONTACT_RECORD_SIZE);
assert(SPIFFS.removes == 0); // first publication has no backup/temp to delete
assert(h.copies >= h.contacts.size());
@@ -515,8 +520,80 @@ void expectedAbsenceAndRealDeleteFailures() {
assert(SPIFFS.removes == 2 && SPIFFS.missing_remove_logs == 0);
}
// Use the production transaction and the same complete 347-contact image for
// old/new stdio admission measurements. Backend calls model fwrite/fread
// buffering, not physical SPIFFS flash operations or hardware latency.
void bufferedAdmissionBenchmark() {
SPIFFS = FakeFilesystem();
SPIFFS.emulate_stdio = true;
const auto bytes = image(table(347));
mesh::ContactFileTransaction writer(&SPIFFS, "/contacts3", companionPathPresence);
assert(writer);
for (size_t offset = 0; offset < bytes.size(); offset += mesh::storage::CONTACT_RECORD_SIZE) {
const auto before = SPIFFS.backend_writes;
assert(writer.write(bytes.data() + offset, mesh::storage::CONTACT_RECORD_SIZE)
== mesh::storage::CONTACT_RECORD_SIZE);
assert(SPIFFS.backend_writes - before <= 1);
}
auto progress = mesh::ContactFileTransaction::CommitProgress::Pending;
unsigned passes = 0;
while (progress == mesh::ContactFileTransaction::CommitProgress::Pending) {
const auto before_reads = SPIFFS.backend_reads, before_writes = SPIFFS.backend_writes;
assert(++passes < 1000);
progress = writer.serviceCommit();
assert(SPIFFS.backend_reads - before_reads <= 1);
assert(SPIFFS.backend_writes - before_writes <= 1);
}
assert(progress == mesh::ContactFileTransaction::CommitProgress::Succeeded);
assert(SPIFFS.files.at("/contacts3") == bytes);
assert(SPIFFS.largest_read == 64);
assert(SPIFFS.largest_write == mesh::storage::CONTACT_RECORD_SIZE);
printf("{\"bytes\":%zu,\"logical_writes\":%u,\"logical_reads\":%u,"
"\"backend_writes\":%u,\"backend_reads\":%u,\"max_backend_write\":%zu,"
"\"max_backend_read\":%zu,\"commit_passes\":%u,\"transaction_size\":%zu}\n",
bytes.size(), SPIFFS.writes, SPIFFS.reads, SPIFFS.backend_writes,
SPIFFS.backend_reads, SPIFFS.largest_backend_write,
SPIFFS.largest_backend_read, passes, sizeof(writer));
}
void boundedCancellationFlush() {
// 251 records land exactly on a 251-byte stdio boundary because the record
// size is 152. Both sides of that boundary must preserve the committed
// image and flush at most one partial buffer when abandoning the temp.
for (unsigned records : {1U, 250U, 251U, 252U}) {
SPIFFS = FakeFilesystem();
SPIFFS.emulate_stdio = true;
const auto old = image(table(1));
const auto bytes = image(table(records));
SPIFFS.files["/contacts3"] = old;
unsigned before_cancel = 0;
{
mesh::ContactFileTransaction writer(&SPIFFS, "/contacts3", companionPathPresence);
assert(writer);
for (size_t offset = 0; offset < bytes.size(); offset += mesh::storage::CONTACT_RECORD_SIZE) {
const auto before = SPIFFS.backend_writes;
assert(writer.write(bytes.data() + offset, mesh::storage::CONTACT_RECORD_SIZE)
== mesh::storage::CONTACT_RECORD_SIZE);
assert(SPIFFS.backend_writes - before <= 1);
}
assert(SPIFFS.backend_writes == bytes.size() / 251);
assert(SPIFFS.files.at("/contacts3") == old);
before_cancel = SPIFFS.backend_writes;
}
assert(SPIFFS.backend_writes - before_cancel == (bytes.size() % 251 != 0));
assert(SPIFFS.backend_writes == (bytes.size() + 250) / 251);
assert(SPIFFS.largest_backend_write <= 251);
assert(SPIFFS.files.at("/contacts3") == old);
assert(!SPIFFS.exists("/contacts3.tmp"));
}
}
int main() {
#if defined(CONTACT_BUFFER_BENCHMARK)
bufferedAdmissionBenchmark();
#else
boundedFullTableAndFirstSave();
boundedCancellationFlush();
deferredBeginCancellationAndCrashRecovery();
deferredBeginFaultsAndSynchronousDefaults();
mutationDuringEveryBeginStage();
@@ -526,4 +603,5 @@ int main() {
realBackupCleanupFailure();
rebootAbortRecoveryAndDurableRemoval();
expectedAbsenceAndRealDeleteFailures();
#endif
}
+52 -6
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
"""Run production ESP contact transactions with inline paths, including PSRAM."""
from pathlib import Path
import json
import subprocess
import tempfile
import unittest
@@ -25,7 +26,28 @@ class UncachedESPContactsTest(unittest.TestCase):
def test_synchronous_startup_batch_is_detected(self):
self.run_fixture(["-DBOARD_HAS_PSRAM=1"], disable="begin")
def run_fixture(self, policy, disable=None):
def test_bounded_buffers_reduce_backend_work_without_stack_growth(self):
previous = self.run_fixture(["-DBOARD_HAS_PSRAM=1"], benchmark="previous")
current = self.run_fixture(["-DBOARD_HAS_PSRAM=1"], benchmark="current")
for result in (previous, current):
self.assertEqual(result["bytes"], 347 * 152)
self.assertEqual(result["logical_writes"], 347)
self.assertEqual(result["logical_reads"], 825)
self.assertEqual(previous["backend_writes"], 347)
self.assertEqual(current["backend_writes"], 211)
self.assertEqual(previous["backend_reads"], 825)
self.assertEqual(current["backend_reads"], 413)
self.assertEqual(current["max_backend_write"], 251)
self.assertEqual(current["max_backend_read"], 128)
self.assertEqual(current["commit_passes"], previous["commit_passes"])
self.assertEqual(current["transaction_size"], previous["transaction_size"])
# Actual production methods are compiled out of line with identical
# options. Buffer changes must not increase their native stack frames.
self.assertEqual(current["stack_frames"], previous["stack_frames"])
self.assertTrue(any("serviceCommit" in method for method in current["stack_frames"]))
self.assertTrue(any("serviceBegin" in method for method in current["stack_frames"]))
def run_fixture(self, policy, disable=None, benchmark=None):
with tempfile.TemporaryDirectory(prefix="mesh-esp-inline-contacts-") as directory:
temp = Path(directory)
source = (ROOT / "examples/companion_radio/DataStore.cpp").read_text()
@@ -71,14 +93,21 @@ class UncachedESPContactsTest(unittest.TestCase):
implementation += method(utils, "void Utils::toHex(") + "\n}\n"
(temp / "store_under_test.h").write_text(implementation)
transaction = (ROOT / "src/helpers/ContactFileTransaction.h").read_text()
if benchmark == "previous":
self.assertIn("ESP_WRITE_BUFFER_SIZE = 251;", transaction)
self.assertIn("ESP_VERIFY_BUFFER_SIZE = 128;", transaction)
transaction = transaction.replace("ESP_WRITE_BUFFER_SIZE = 251;",
"ESP_WRITE_BUFFER_SIZE = 152;")
transaction = transaction.replace("ESP_VERIFY_BUFFER_SIZE = 128;",
"ESP_VERIFY_BUFFER_SIZE = 64;")
if disable == "writer":
self.assertIn("ok = _file.setBufferSize(storage::CONTACT_RECORD_SIZE);", transaction)
self.assertIn("ok = _file.setBufferSize(ESP_WRITE_BUFFER_SIZE);", transaction)
transaction = transaction.replace(
"ok = _file.setBufferSize(storage::CONTACT_RECORD_SIZE);",
"ok = _file.setBufferSize(ESP_WRITE_BUFFER_SIZE);",
"ok = true; // Negative control: retain default writer buffer.")
elif disable == "reader":
self.assertIn("ok = ok && _verify.setBufferSize(64);", transaction)
transaction = transaction.replace("ok = ok && _verify.setBufferSize(64);",
self.assertIn("ok = ok && _verify.setBufferSize(ESP_VERIFY_BUFFER_SIZE);", transaction)
transaction = transaction.replace("ok = ok && _verify.setBufferSize(ESP_VERIFY_BUFFER_SIZE);",
"// Negative control: retain the SDK's default reader buffer.")
transaction = transaction.replace('#include "IdentityStore.h"',
'#include <helpers/IdentityStore.h>')
@@ -96,10 +125,12 @@ class UncachedESPContactsTest(unittest.TestCase):
"uint8_t Packet::copyPath("))
+ "\n}\n")
binary = temp / "test"
benchmark_options = (["-DCONTACT_BUFFER_BENCHMARK=1", "-fstack-usage", "-fno-inline"]
if benchmark else [])
result = subprocess.run([
"c++", "-std=c++17", "-O1", "-g", "-Wall", "-Wextra",
"-fsanitize=address,undefined", "-fno-omit-frame-pointer",
"-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1", *policy,
"-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1", *policy, *benchmark_options,
"-I", str(ROOT / "test/fixtures/contact_cache/mocks"),
"-I", str(ROOT / "test/fixtures/contact_cache"),
"-I", str(ROOT / "lib/ed25519"), "-I", str(ROOT / "src"),
@@ -111,6 +142,21 @@ class UncachedESPContactsTest(unittest.TestCase):
result = subprocess.run([str(binary)], capture_output=True, text=True)
if disable is None:
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
if benchmark:
measurements = json.loads(result.stdout)
frames = {}
for usage in temp.glob("*.su"):
for line in usage.read_text().splitlines():
fields = line.split("\t")
if len(fields) == 3 and "ContactFileTransaction::" in fields[0]:
# Strip the generated path/line prefix; all
# method frames (including CRC verification)
# must remain unchanged between both builds.
name = fields[0].split("mesh::ContactFileTransaction", 1)[1]
frames[name] = [int(fields[1]), fields[2]]
self.assertTrue(frames)
measurements["stack_frames"] = frames
return measurements
else:
self.assertNotEqual(result.returncode, 0)
marker = "startup_operations" if disable == "begin" else (