From 4939de4c61b868cd56d592232696143c68521134 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Sun, 4 Oct 2026 21:40:43 -0700 Subject: [PATCH] Keep hardware CDC packet log lines whole under congestion --- src/helpers/SerialPacketLog.h | 12 ++++ src/helpers/UsbLogging.cpp | 7 ++- test/fixtures/hwcdc_write_capacity.cpp | 87 ++++++++++++++++++++++++++ test/test_esp32_usb_serial_hygiene.py | 11 +++- test/test_hwcdc_write_capacity.py | 42 ++++++++++--- 5 files changed, 150 insertions(+), 9 deletions(-) diff --git a/src/helpers/SerialPacketLog.h b/src/helpers/SerialPacketLog.h index 373657c0..4a5d964b 100644 --- a/src/helpers/SerialPacketLog.h +++ b/src/helpers/SerialPacketLog.h @@ -32,6 +32,17 @@ inline bool& serialLogPortSeen() { template bool serialLogEmit(T& out, const char* data, size_t len) { +#if defined(ESP32_PLATFORM) && defined(ARDUINO_USB_CDC_ON_BOOT) && \ + (ARDUINO_USB_CDC_ON_BOOT == 1) && defined(ARDUINO_USB_MODE) && \ + (ARDUINO_USB_MODE == 1) + // The HWCDC logging view checks this complete record again under the shared + // producer guard. Never leave a line prefix behind when its host stalls. + const int space = out.availableForWrite(); + if (space <= 0 || static_cast(space) < len) return false; + if (out.write(reinterpret_cast(data), len) != len) return false; + serialLogPortSeen() = true; + return true; +#else if (out.availableForWrite() <= 0) return false; const uint32_t start = millis(); @@ -55,6 +66,7 @@ bool serialLogEmit(T& out, const char* data, size_t len) { } serialLogPortSeen() = true; return true; +#endif } template diff --git a/src/helpers/UsbLogging.cpp b/src/helpers/UsbLogging.cpp index 43af9bf4..3ba67065 100644 --- a/src/helpers/UsbLogging.cpp +++ b/src/helpers/UsbLogging.cpp @@ -513,6 +513,11 @@ static SingleAttemptNonBlockingStream guarded_esp32_hwcdc_port( static AtomicWholeRecordNonBlockingStream<11> guarded_esp32_hwcdc_mota_port(guarded_esp32_hwcdc_port); +// Admit each complete packet log line under the same guard as CLI replies. +// Short records can still use a smaller ring if the preferred allocation fails. +static AtomicWholeRecordNonBlockingStream<640> guarded_esp32_hwcdc_logging_port( + guarded_esp32_hwcdc_port); + static void serviceEsp32HwcdcTxKickExclusive(void*) { if (!esp32_hwcdc_tx_kick_pending.load(std::memory_order_acquire) || !canAccessEsp32Hwcdc(nullptr)) { @@ -1393,7 +1398,7 @@ Stream& usbLoggingPort() { #if MESH_ESP32_TINYUSB_NONBLOCKING return buffered_esp32_tinyusb_logging_port; #elif MESH_ESP32_HWCDC_SESSION_GUARD - return guarded_esp32_hwcdc_port; + return guarded_esp32_hwcdc_logging_port; #elif defined(NRF52_PLATFORM) return nonblocking_primary_usb_logging_port; #else diff --git a/test/fixtures/hwcdc_write_capacity.cpp b/test/fixtures/hwcdc_write_capacity.cpp index 0dfb77db..f61432e5 100644 --- a/test/fixtures/hwcdc_write_capacity.cpp +++ b/test/fixtures/hwcdc_write_capacity.cpp @@ -88,6 +88,8 @@ static size_t once(void*, const uint8_t* bytes, size_t size) { } static bool canAccess(void*) { return access_allowed; } static SingleAttemptNonBlockingStream guarded(session, once, nullptr, canAccess); +@LOGGING_VIEW@ +static LoggingView logging(guarded); } static const uint8_t bytes[] = "abcdefghijklmnopqrstuvwx"; @@ -179,6 +181,91 @@ int main(int argc, char** argv) { ring_free = 5; assert(Serial.driverWrite(bytes, 24) == 5); assert(g_mock_millis == 5 && zero_sends == 5 && !connected); + } else if (name == "logging_low") { + const std::vector line(640, 'L'); + ring_free = 639; + assert(!mesh::serialLogEmit(mesh::logging, + reinterpret_cast(line.data()), line.size())); + assert(admitted.empty() && sdk_calls == 0 && ring_sends == 0); + assert(g_mock_millis == 0 && zero_sends == 0); + } else if (name == "logging_full") { + const std::vector line(640, 'L'); + ring_free = 640; + assert(mesh::logging.availableForWrite() == 640); + assert(mesh::serialLogEmit(mesh::logging, + reinterpret_cast(line.data()), line.size())); + assert(admitted == line && sdk_calls == 1 && ring_sends == 1); + assert(g_mock_millis == 0 && zero_sends == 0); + } else if (name == "logging_fallback_small") { + const std::vector line(100, 'L'); + ring_free = 256; + assert(mesh::logging.availableForWrite() == 256); + assert(mesh::serialLogEmit(mesh::logging, + reinterpret_cast(line.data()), line.size())); + assert(admitted == line && sdk_calls == 1 && ring_sends == 1); + assert(g_mock_millis == 0 && zero_sends == 0); + } else if (name == "logging_fallback_large") { + const std::vector line(257, 'L'); + ring_free = 256; + assert(!mesh::serialLogEmit(mesh::logging, + reinterpret_cast(line.data()), line.size())); + assert(admitted.empty() && sdk_calls == 0 && ring_sends == 0); + assert(g_mock_millis == 0 && zero_sends == 0); + } else if (name == "logging_stale") { + struct StaleView : Stream { + int availableForWrite() override { return mesh::logging.availableForWrite(); } + size_t write(const uint8_t* data, size_t size) override { + // A different facade uses capacity after serialLogEmit's outer sample. + assert(mesh::guarded.write(bytes, 24) == 24); + return mesh::logging.write(data, size); + } + } stale; + const std::vector line(640, 'L'); + ring_free = 640; + assert(!mesh::serialLogEmit(stale, + reinterpret_cast(line.data()), line.size())); + assert(admitted == std::vector(bytes, bytes + 24)); + assert(sdk_calls == 1 && ring_sends == 1 && g_mock_millis == 0); + } else if (name == "logging_line") { + ring_free = 640; + mesh::SerialLogLine<> line; + line.printf("STREAM_RX: "); + line.hex(bytes, 24); + assert(line.flush(mesh::logging, false)); + const std::string expected = + "STREAM_RX: 6162636465666768696A6B6C6D6E6F707172737475767778\n"; + assert(admitted == std::vector(expected.begin(), expected.end())); + assert(sdk_calls == 1 && ring_sends == 1 && g_mock_millis == 0); + } else if (name == "logging_contention") { + ring_free = 640; + const std::vector first(640, 'A'), second(640, 'B'); + std::mutex mutex; + std::condition_variable event; + bool entered = false, proceed = false; + before_sdk_write = [&] { + std::unique_lock lock(mutex); + entered = true; + event.notify_all(); + assert(event.wait_for(lock, std::chrono::seconds(2), [&] { return proceed; })); + }; + std::thread writer([&] { + assert(mesh::serialLogEmit(mesh::logging, + reinterpret_cast(first.data()), first.size())); + }); + { + std::unique_lock lock(mutex); + assert(event.wait_for(lock, std::chrono::seconds(2), [&] { return entered; })); + } + assert(!mesh::serialLogEmit(mesh::logging, + reinterpret_cast(second.data()), second.size())); + { + std::lock_guard lock(mutex); + proceed = true; + } + event.notify_all(); + writer.join(); + assert(admitted == first && sdk_calls == 1 && ring_sends == 1); + assert(g_mock_millis == 0 && zero_sends == 0); } else { assert(false && "unknown fixture case"); } diff --git a/test/test_esp32_usb_serial_hygiene.py b/test/test_esp32_usb_serial_hygiene.py index a2a12f52..4314c029 100644 --- a/test/test_esp32_usb_serial_hygiene.py +++ b/test/test_esp32_usb_serial_hygiene.py @@ -354,7 +354,16 @@ int main() { self.assertIn( "esp32_hwcdc_tx_kick_pending.store(true", guarded_write ) - self.assertIn("return Serial.write(data, size);", guarded_write) + self.assertIn("return Serial.write(data, attempt);", guarded_write) + capacity = guarded_write.index("const int available = Serial.availableForWrite();") + full = guarded_write.index("if (available <= 0) return 0;") + clamp = guarded_write.index("const size_t attempt = size <") + kick_pending = guarded_write.index("esp32_hwcdc_tx_kick_pending.store(true") + native_write = guarded_write.index("return Serial.write(data, attempt);") + self.assertLess(capacity, full) + self.assertLess(full, clamp) + self.assertLess(clamp, kick_pending) + self.assertLess(kick_pending, native_write) # Extract complete functions: their pinned-SDK/fallback branches now # contain nested #endif directives inside the outer platform guard. diff --git a/test/test_hwcdc_write_capacity.py b/test/test_hwcdc_write_capacity.py index 30edba99..f3b2e1a5 100644 --- a/test/test_hwcdc_write_capacity.py +++ b/test/test_hwcdc_write_capacity.py @@ -2,6 +2,7 @@ """Execute the HWCDC facade and pinned SDK write against a stalled host.""" from pathlib import Path import os +import re import shutil import subprocess import tempfile @@ -26,12 +27,17 @@ class HwcdcWriteCapacityTest(unittest.TestCase): sdk_write = body(PATCHED, 'size_t HWCDC::write(const uint8_t *buffer, size_t size)') sdk_write = sdk_write.replace('HWCDC::write(', 'HWCDC::driverWrite(', 1) sdk_available = body(PATCHED, 'int HWCDC::availableForWrite(void)') + logging = re.search(r'static\s+(AtomicWholeRecordNonBlockingStream<640>)\s+' + r'guarded_esp32_hwcdc_logging_port\(', source) + self.assertIsNotNone(logging, 'use the actual HWCDC logging view type') return (FIXTURE.read_text() .replace('@SDK_WRITE@', sdk_write) .replace('@SDK_AVAILABLE@', sdk_available) - .replace('@FACADE@', facade)) + .replace('@FACADE@', facade) + .replace('@LOGGING_VIEW@', 'using LoggingView = ' + logging[1] + ';')) - def compile(self, directory, uncapped=False, zero_timeout=False): + def compile(self, directory, uncapped=False, zero_timeout=False, + partial_logging=False): compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++') if compiler is None: self.skipTest('a host C++17 compiler is required') @@ -40,14 +46,26 @@ class HwcdcWriteCapacityTest(unittest.TestCase): flags = ['-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie'] if os.name != 'nt' else [] includes = ['-I' + str(ROOT / 'test/mocks'), '-I' + str(ROOT / 'src')] - if zero_timeout: + if zero_timeout or partial_logging: copied = Path(directory) / 'helpers' copied.mkdir() packet_log = (ROOT / 'src/helpers/SerialPacketLog.h').read_text() - self.assertIn('Serial.setTxTimeoutMs(5);', packet_log) + if zero_timeout: + self.assertIn('Serial.setTxTimeoutMs(5);', packet_log) + packet_log = packet_log.replace('Serial.setTxTimeoutMs(5);', + 'Serial.setTxTimeoutMs(0);') + if partial_logging: + # Restore the actual old non-HWCDC chunked emitter, preserving + # the rest of this copied production header byte-for-byte. + function = body(packet_log, 'bool serialLogEmit(') + start = function.index('#if defined(ESP32_PLATFORM)') + old_branch = function.index('\n#else\n') + end = function.rindex('\n#endif') + restored = (function[:start] + + function[old_branch + len('\n#else\n'):end] + '\n}') + packet_log = packet_log.replace(function, restored) (copied / 'SerialPacketLog.h').write_text( - packet_log.replace('Serial.setTxTimeoutMs(5);', - 'Serial.setTxTimeoutMs(0);'), encoding='ascii') + packet_log, encoding='ascii') includes.insert(0, '-I' + str(directory)) binary = Path(directory) / 'test.exe' built = subprocess.run([ @@ -62,7 +80,10 @@ class HwcdcWriteCapacityTest(unittest.TestCase): with tempfile.TemporaryDirectory(prefix='hwcdc-capacity-') as directory: binary = self.compile(directory) for case in ('full', 'oversize', 'stale', 'contention', 'mutex', - 'isr_drain', 'short', 'mota', 'timeout', 'sdk_fallback'): + 'isr_drain', 'short', 'mota', 'timeout', 'sdk_fallback', + 'logging_low', 'logging_full', 'logging_line', 'logging_stale', + 'logging_contention', 'logging_fallback_small', + 'logging_fallback_large'): with self.subTest(case=case): result = subprocess.run([str(binary), case], capture_output=True, text=True, timeout=10) @@ -82,6 +103,13 @@ class HwcdcWriteCapacityTest(unittest.TestCase): text=True, timeout=10) self.assertNotEqual(result.returncode, 0) + def test_partial_logging_capacity_negative_control(self): + with tempfile.TemporaryDirectory(prefix='hwcdc-partial-log-') as directory: + binary = self.compile(directory, partial_logging=True) + result = subprocess.run([str(binary), 'logging_low'], capture_output=True, + text=True, timeout=10) + self.assertNotEqual(result.returncode, 0) + if __name__ == '__main__': unittest.main(verbosity=2)