From ebd40d1e925780d4e34fcdfad007aa2331bdd4f0 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 16 Oct 2026 03:14:37 -0700 Subject: [PATCH] Bound Ethernet receive work to the entry byte budget --- .../ethernet/SerialEthernetInterface.cpp | 9 ++- .../serial_ethernet_sessions/test.cpp | 56 +++++++++++++++ test/test_serial_ethernet_sessions.py | 69 ++++++++++++++++--- 3 files changed, 123 insertions(+), 11 deletions(-) diff --git a/src/helpers/ethernet/SerialEthernetInterface.cpp b/src/helpers/ethernet/SerialEthernetInterface.cpp index be788f0f6..b13b53378 100644 --- a/src/helpers/ethernet/SerialEthernetInterface.cpp +++ b/src/helpers/ethernet/SerialEthernetInterface.cpp @@ -165,10 +165,12 @@ size_t SerialEthernetInterface::checkRecvFrame(uint8_t dest[]) { send_queue[i] = send_queue[i + 1]; } } else { - while (true) { + // Newly arriving bytes belong to the next mesh loop. A continuously + // writing peer must not monopolize radio and watchdog service. + int pending = available(); + while (pending-- > 0) { // A read may likewise hand its receive-ACK command to the worker; // leave the rest of the frame untouched until hardware is available. - if (updateIOPause((uint32_t)millis()) || !available()) break; if (updateIOPause((uint32_t)millis())) break; int c = read(); if (c < 0) break; @@ -231,6 +233,9 @@ size_t SerialEthernetInterface::checkRecvFrame(uint8_t dest[]) { } #endif } + // Snapshot availability or the final budgeted read can hand IO to the + // worker. Observe that pause after the byte has changed parser state. + (void)updateIOPause((uint32_t)millis()); } } diff --git a/test/fixtures/serial_ethernet_sessions/test.cpp b/test/fixtures/serial_ethernet_sessions/test.cpp index 8f620664b..372059cb9 100644 --- a/test/fixtures/serial_ethernet_sessions/test.cpp +++ b/test/fixtures/serial_ethernet_sessions/test.cpp @@ -42,6 +42,9 @@ struct Transport : SerialEthernetInterface { bool pause_after_write = false; bool pause_after_read = false; bool pause_after_available = false; + bool replenish_read = false; + size_t read_guard = 0; + Bytes append_after_read; int available_calls = 0; int write_calls = 0; int disconnect_calls = 0; @@ -67,9 +70,16 @@ struct Transport : SerialEthernetInterface { int read() override { assert(!io_unavailable); if (!client || client->received.empty()) return -1; + assert((read_guard == 0 || client->read_count < read_guard) + && "Ethernet RX invocation exceeded bounded read guard"); const int value = client->received.front(); client->received.pop_front(); ++client->read_count; + if (replenish_read) client->received.push_back(uint8_t(value)); + if (!append_after_read.empty()) { + client->received.insert(client->received.end(), append_after_read.begin(), append_after_read.end()); + append_after_read.clear(); + } if (pause_after_read) { io_unavailable = true; pause_after_read = false; @@ -500,6 +510,52 @@ int main(int argc, char** argv) { assert(f.tick() == 0 && !socket->connected && f.callbacks == 1); #endif assert(other_socket->connected); + } else if (scenario >= 30 && scenario <= 35) { + const bool replenish = scenario <= 32; + static constexpr size_t budgets[] = {1, 7, 257}; + const size_t budget = budgets[(scenario - 30) % 3]; + const Bytes request = command(Bytes{'a', 'b'}); +#if ETHERNET_RAW_LINE + const uint8_t junk = '\n'; +#else + const uint8_t junk = 'x'; +#endif + socket->received.assign(budget, junk); + f.transport.read_guard = 4096; // The live-availability negative control must terminate. + f.transport.replenish_read = replenish; + if (!replenish) f.transport.append_after_read = request; + assert(f.tick() == 0 && "new arrivals must wait for the next mesh loop"); + assert(socket->read_count == budget && f.transport.available_calls == 1); + assert(socket->received.size() == (replenish ? budget : request.size())); + assert(socket->connected && !f.transport.isReadBusy() && f.callbacks == 0); + f.transport.replenish_read = false; + if (replenish) socket->received.insert(socket->received.end(), request.begin(), request.end()); + // Nothing was dropped: existing idle bytes and a newly arrived complete + // command remain in wire order and parse on the following mesh loop. + assert(f.tick() == 2 && f.input[0] == 'a' && f.input[1] == 'b'); + assert(socket->received.empty() && f.transport.available_calls == 2); + } else if (scenario >= 36 && scenario <= 39) { + const bool zero_available = scenario >= 38; + mock_millis = (scenario & 1) ? UINT32_MAX - 1000U : 0; + const Bytes request = command(Bytes{'a', 'b'}); + socket->received.push_back(request[0]); + if (!zero_available) f.transport.pause_after_read = true; + assert(f.tick() == 0 && f.transport.isReadBusy()); + if (zero_available) { + mock_millis += 2000; + f.transport.pause_after_available = true; + assert(f.tick() == 0); + } + assert(f.transport.io_unavailable && socket->read_count == 1); + const int available_calls = f.transport.available_calls; + mock_millis += 12000; + assert(f.tick() == 0 && f.transport.available_calls == available_calls); + f.transport.io_unavailable = false; + f.transport.loop(); + mock_millis += zero_available ? 2999 : 4999; + socket->received.insert(socket->received.end(), request.begin() + 1, request.end()); + assert((f.tick() == 2 && socket->connected && f.input[0] == 'a' && f.input[1] == 'b' + && f.callbacks == 0) && "worker handoff must preserve remaining frame deadline"); } else { assert(false && "unknown scenario"); } diff --git a/test/test_serial_ethernet_sessions.py b/test/test_serial_ethernet_sessions.py index 4a56d7b86..08a4a827d 100644 --- a/test/test_serial_ethernet_sessions.py +++ b/test/test_serial_ethernet_sessions.py @@ -46,10 +46,30 @@ SCENARIOS = ( "read command ownership pauses before polling the socket again", "availability command ownership prevents a following socket read", "main-loop ownership service preserves timeout behind another locked route", + "continuously replenished idle input yields after entry budget 1", + "continuously replenished idle input yields after entry budget 7", + "continuously replenished idle input yields after entry budget 257", + "new command waits behind entry budget 1 until next loop", + "new command waits behind entry budget 7 until next loop", + "new command waits behind entry budget 257 until next loop", + "last budgeted read records worker pause after parser state changes", + "last budgeted read worker pause preserves deadline across rollover", + "zero-available snapshot records worker pause immediately", + "zero-available worker pause preserves deadline across rollover", ) class SerialEthernetSessionTests(unittest.TestCase): + def compile_transport(self, compiler, raw, binary, source): + compiled = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + *SANITIZER_FLAGS, f"-DETHERNET_RAW_LINE={raw}", + f"-I{MOCKS}", f"-I{ROOT / 'src'}", + f"-I{ROOT / 'src/helpers/ethernet'}", str(FIXTURE / "test.cpp"), + str(source), "-o", str(binary)], capture_output=True, text=True, timeout=60, + ) + self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr) + def test_production_transport_sessions(self): compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") if compiler is None: @@ -57,15 +77,8 @@ class SerialEthernetSessionTests(unittest.TestCase): with tempfile.TemporaryDirectory(prefix="meshcore-ethernet-sessions-") as directory: binary = Path(directory) / "ethernet-sessions.exe" for raw in (0, 1): - compiled = subprocess.run( - [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", - *SANITIZER_FLAGS, f"-DETHERNET_RAW_LINE={raw}", - f"-I{MOCKS}", f"-I{ROOT / 'src'}", - str(FIXTURE / "test.cpp"), - str(ROOT / "src/helpers/ethernet/SerialEthernetInterface.cpp"), - "-o", str(binary)], capture_output=True, text=True, timeout=60, - ) - self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr) + self.compile_transport(compiler, raw, binary, + ROOT / "src/helpers/ethernet/SerialEthernetInterface.cpp") for scenario, description in enumerate(SCENARIOS): with self.subTest(raw=raw, scenario=description): checked = subprocess.run( @@ -75,6 +88,44 @@ class SerialEthernetSessionTests(unittest.TestCase): self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) self.assertIn(f"PASS: Ethernet session scenario {scenario}", checked.stdout) + def test_live_available_and_missing_final_pause_negative_controls(self): + 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 = (ROOT / "src/helpers/ethernet/SerialEthernetInterface.cpp").read_text() + snapshot = ''' int pending = available(); + while (pending-- > 0) { + // A read may likewise hand its receive-ACK command to the worker; + // leave the rest of the frame untouched until hardware is available. + if (updateIOPause((uint32_t)millis())) break;''' + live = ''' while (true) { + if (updateIOPause((uint32_t)millis()) || !available()) break; + if (updateIOPause((uint32_t)millis())) break;''' + pause = ''' // Snapshot availability or the final budgeted read can hand IO to the + // worker. Observe that pause after the byte has changed parser state. + (void)updateIOPause((uint32_t)millis());''' + self.assertEqual(source.count(snapshot), 1) + self.assertEqual(source.count(pause), 1) + with tempfile.TemporaryDirectory(prefix="meshcore-ethernet-rx-negative-") as directory: + work = Path(directory) + for name, modified, modes, scenarios in ( + ("live", source.replace(snapshot, live), (0, 1), range(30, 36)), + ("naive", source.replace(pause, ""), (0,), range(36, 40))): + variant = work / (name + ".cpp") + variant.write_text(modified) + for raw in modes: + binary = work / (name + str(raw) + ".exe") + self.compile_transport(compiler, raw, binary, variant) + for scenario in scenarios: + with self.subTest(variant=name, raw=raw, scenario=SCENARIOS[scenario]): + checked = subprocess.run([str(binary), str(scenario)], capture_output=True, + text=True, timeout=10) + self.assertNotEqual(checked.returncode, 0, "unsafe RX loop escaped regression") + expected = ("bounded read guard" if scenario <= 32 else + "new arrivals must wait" if scenario <= 35 else + "worker handoff must preserve remaining frame deadline") + self.assertIn(expected, checked.stderr) + def test_hardware_drivers_cancel_before_exposing_new_client(self): for driver in ("ch390/CH390EthernetInterface", "RAK13800/RAK13800EthernetInterface"): source = (ROOT / f"src/helpers/ethernet/{driver}.cpp").read_text(encoding="utf-8")