diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index dc3582db..0c70c4fc 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -266,6 +266,7 @@ jobs: python3 -B test/test_esp32_tinyusb_nonblocking.py python3 -B test/test_esp32_usb_session_fix.py python3 -B test/test_hwcdc_tx_backport.py -v + python3 -B test/test_hwcdc_write_capacity.py -v python3 -B test/test_esp32_file_buffer_fix.py -v python3 -B test/test_companion_usb_logging_reply.py python3 -B test/test_esp32_tinyusb_cooperative_output.py diff --git a/test/fixtures/hwcdc_write_capacity.cpp b/test/fixtures/hwcdc_write_capacity.cpp new file mode 100644 index 00000000..0dfb77db --- /dev/null +++ b/test/fixtures/hwcdc_write_capacity.cpp @@ -0,0 +1,189 @@ +// The Python runner inserts the actual facade and pinned, patched SDK methods. +// Only the RTOS mutex/ring and hardware registers are replaced at this boundary. +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static size_t ring_free = 0; +static std::vector admitted; +static unsigned ring_sends = 0, zero_sends = 0, sdk_calls = 0; +static unsigned semaphore_takes = 0, tx_attempts = 0; +static bool reject_send = false, mutex_available = true; +static bool connected = true, access_allowed = true; +static uint32_t tx_timeout_ms = 99; +static int ring_token, lock_token; +static int* tx_ring_buf = &ring_token; +static int* tx_lock = &lock_token; +static std::function before_sdk_write; +constexpr int pdPASS = 1, pdTRUE = 1; +constexpr unsigned portTICK_PERIOD_MS = 1; +static int xSemaphoreTake(int*, unsigned ticks) { + ++semaphore_takes; + if (mutex_available) return pdPASS; + delay(ticks); + return 0; +} +static void xSemaphoreGive(int*) {} +static size_t xRingbufferGetCurFreeSize(int*) { return ring_free; } +static int xRingbufferSend(int*, void* data, size_t size, unsigned) { + ++ring_sends; + if (size == 0) { ++zero_sends; return pdTRUE; } + if (reject_send || size > ring_free) return 0; + const auto* bytes = static_cast(data); + admitted.insert(admitted.end(), bytes, bytes + size); + ring_free -= size; + return pdTRUE; +} +static void usb_serial_jtag_ll_txfifo_flush() {} +static bool usb_serial_jtag_ll_txfifo_writable() { return true; } +static void mesh_hwcdc_enable_tx_intr() {} +static void meshEsp32HwcdcKickTx() {} +static void log_w(const char*) {} +struct mesh_hwcdc_writer_scope { explicit operator bool() const { return true; } }; +class HWCDC : public Stream { + public: + static bool isCDC_Connected() { return connected; } + int available() override { return 0; } + int read() override { return -1; } + int peek() override { return -1; } + void flush() override { assert(false && "guard must suppress flush"); } + int availableForWrite() override; + size_t driverWrite(const uint8_t*, size_t); + size_t write(const uint8_t* bytes, size_t size) override { + ++sdk_calls; + if (before_sdk_write) { + auto callback = std::move(before_sdk_write); + before_sdk_write = nullptr; + callback(); + } + return driverWrite(bytes, size); + } + void setTxTimeoutMs(uint32_t timeout) { tx_timeout_ms = timeout; } + void flushTXBuffer(const uint8_t* bytes, size_t size) { + assert(size <= ring_free); + assert(xRingbufferSend(tx_ring_buf, const_cast(bytes), size, 0) == pdTRUE); + } +} Serial; +@SDK_AVAILABLE@ +@SDK_WRITE@ +#include +namespace mesh { +static UsbSelfResetBurstGuard esp32_hwcdc_self_reset_guard; +static std::atomic esp32_hwcdc_tx_kick_pending{false}; +static void noteUsbLoggingTxAttempt() { ++tx_attempts; } +@FACADE@ +static Esp32HwcdcSessionStream session; +static size_t once(void*, const uint8_t* bytes, size_t size) { + return session.write(bytes, size); +} +static bool canAccess(void*) { return access_allowed; } +static SingleAttemptNonBlockingStream guarded(session, once, nullptr, canAccess); +} +static const uint8_t bytes[] = "abcdefghijklmnopqrstuvwx"; + +int main(int argc, char** argv) { + assert(argc == 2); + const std::string name = argv[1]; + using mesh::guarded; + mesh::serialLogBegin(); + assert(tx_timeout_ms == 5); + if (name == "full") { + ring_free = 0; + assert(guarded.write(bytes, 24) == 0); + assert(sdk_calls == 0 && ring_sends == 0 && tx_attempts == 0); + } else if (name == "oversize") { + ring_free = 5; + assert(guarded.write(bytes, 24) == 5); + assert(g_mock_millis == 0 && zero_sends == 0 && ring_sends == 1); + } else if (name == "stale") { + ring_free = 12; + const int prior = guarded.availableForWrite(); + assert(guarded.write(bytes, 7) == 7); + assert(guarded.write(bytes + 7, prior) == 5); + assert(g_mock_millis == 0 && zero_sends == 0); + assert(admitted == std::vector(bytes, bytes + 12)); + } else if (name == "contention") { + ring_free = 24; + 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(guarded.write(bytes, 7) == 7); }); + { + std::unique_lock lock(mutex); + assert(event.wait_for(lock, std::chrono::seconds(2), [&] { return entered; })); + } + assert(guarded.write(bytes + 7, 7) == 0); + assert(!guarded.tryRunExclusive([](void*) { assert(false); })); + { + std::lock_guard lock(mutex); + proceed = true; + } + event.notify_all(); + writer.join(); + assert(sdk_calls == 1 && admitted.size() == 7 && g_mock_millis == 0); + } else if (name == "mutex") { + ring_free = 24; + mutex_available = false; + assert(guarded.write(bytes, 24) == 0); + assert(sdk_calls == 0 && semaphore_takes == 1 && g_mock_millis == 5); + } else if (name == "isr_drain") { + ring_free = 5; + before_sdk_write = [] { ring_free += 8; }; + assert(guarded.write(bytes, 24) == 5); + assert(ring_free == 8 && g_mock_millis == 0 && zero_sends == 0); + } else if (name == "short") { + ring_free = 5; + reject_send = true; + assert(guarded.write(bytes, 24) == 0); + reject_send = false; + mesh::BufferedNonBlockingWriteStream<32> queued(guarded); + assert(queued.write(bytes, 24) == 24); + assert(queued.queuedByteCount() == 19 && admitted.size() == 5); + ring_free = 7; + assert(queued.service() == 7); + ring_free = 12; + assert(queued.service() == 12 && queued.queuedByteCount() == 0); + assert(admitted == std::vector(bytes, bytes + 24)); + assert(g_mock_millis == 0 && zero_sends == 0); + } else if (name == "mota") { + mesh::AtomicWholeRecordNonBlockingStream<11> mota(guarded); + ring_free = 10; + assert(mota.write(bytes, 11) == 0 && sdk_calls == 0); + ring_free = 11; + assert(mota.write(bytes, 11) == 11 && admitted.size() == 11); + assert(mota.write(bytes, 12) == 0 && sdk_calls == 1); + assert(g_mock_millis == 0 && zero_sends == 0); + } else if (name == "timeout") { + Serial.setTxTimeoutMs(0); + mesh::serialLogBegin(); + assert(tx_timeout_ms == 5); + } else if (name == "sdk_fallback") { + // This intentionally bypasses the cap to exercise the SDK's positive + // timeout fallback; the application path above never needs this loop. + ring_free = 5; + assert(Serial.driverWrite(bytes, 24) == 5); + assert(g_mock_millis == 5 && zero_sends == 5 && !connected); + } else { + assert(false && "unknown fixture case"); + } + guarded.flush(); + access_allowed = false; + assert(guarded.write(bytes, 1) == 0); + return 0; +} diff --git a/test/test_hwcdc_write_capacity.py b/test/test_hwcdc_write_capacity.py new file mode 100644 index 00000000..30edba99 --- /dev/null +++ b/test/test_hwcdc_write_capacity.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +"""Execute the HWCDC facade and pinned SDK write against a stalled host.""" +from pathlib import Path +import os +import shutil +import subprocess +import tempfile +import unittest + +from test_hwcdc_tx_backport import PATCHED, body + + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = ROOT / 'test/fixtures/hwcdc_write_capacity.cpp' + + +class HwcdcWriteCapacityTest(unittest.TestCase): + def harness(self, uncapped=False): + source = (ROOT / 'src/helpers/UsbLogging.cpp').read_text() + facade = body(source, 'class Esp32HwcdcSessionStream') + ';' + if uncapped: + start = facade.index(' // The shared producer guard') + end = facade.index(' noteUsbLoggingTxAttempt();', start) + facade = facade[:start] + facade[end:] + facade = facade.replace('Serial.write(data, attempt)', 'Serial.write(data, size)') + 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)') + return (FIXTURE.read_text() + .replace('@SDK_WRITE@', sdk_write) + .replace('@SDK_AVAILABLE@', sdk_available) + .replace('@FACADE@', facade)) + + def compile(self, directory, uncapped=False, zero_timeout=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') + source = Path(directory) / 'test.cpp' + source.write_text(self.harness(uncapped), encoding='ascii') + 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: + copied = Path(directory) / 'helpers' + copied.mkdir() + packet_log = (ROOT / 'src/helpers/SerialPacketLog.h').read_text() + self.assertIn('Serial.setTxTimeoutMs(5);', packet_log) + (copied / 'SerialPacketLog.h').write_text( + packet_log.replace('Serial.setTxTimeoutMs(5);', + 'Serial.setTxTimeoutMs(0);'), encoding='ascii') + includes.insert(0, '-I' + str(directory)) + binary = Path(directory) / 'test.exe' + built = subprocess.run([ + compiler, '-std=c++17', '-Wall', '-Wextra', '-Wno-unused-parameter', + '-pthread', *flags, '-DESP32_PLATFORM', '-DARDUINO_USB_MODE=1', + '-DARDUINO_USB_CDC_ON_BOOT=1', *includes, str(source), '-o', str(binary) + ], capture_output=True, text=True, timeout=60) + self.assertEqual(built.returncode, 0, built.stderr) + return binary + + def test_actual_facade_sdk_stall_capacity_and_record_boundaries(self): + 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'): + with self.subTest(case=case): + result = subprocess.run([str(binary), case], capture_output=True, + text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_uncapped_negative_control_reenters_sdk_wait_loop(self): + with tempfile.TemporaryDirectory(prefix='hwcdc-no-cap-') as directory: + binary = self.compile(directory, uncapped=True) + result = subprocess.run([str(binary), 'oversize'], capture_output=True, + text=True, timeout=10) + self.assertNotEqual(result.returncode, 0) + + def test_zero_timeout_negative_control(self): + with tempfile.TemporaryDirectory(prefix='hwcdc-no-timeout-') as directory: + binary = self.compile(directory, zero_timeout=True) + result = subprocess.run([str(binary), 'timeout'], capture_output=True, + text=True, timeout=10) + self.assertNotEqual(result.returncode, 0) + + +if __name__ == '__main__': + unittest.main(verbosity=2)