From babb06923701d5468369f5c6aaed93f1850c4c0a Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 16 Oct 2026 05:03:34 -0700 Subject: [PATCH] Prevent pipelined clients from starving Companion transports --- src/helpers/MultiSerialInterface.h | 11 +- .../fixtures/companion_wifi_listener/test.cpp | 173 ++++++++++++++++++ test/test_companion_wifi_listener.py | 41 +++++ 3 files changed, 223 insertions(+), 2 deletions(-) diff --git a/src/helpers/MultiSerialInterface.h b/src/helpers/MultiSerialInterface.h index 5ea761525..d5a4ec8c1 100644 --- a/src/helpers/MultiSerialInterface.h +++ b/src/helpers/MultiSerialInterface.h @@ -28,6 +28,7 @@ private: bool _enabled = false; bool _bluetooth_auto_enable = true; + uint8_t _nextRxIndex = 0; RegisteredInterface _interfaces[MAX_INTERFACES] = {}; BaseSerialInterface* _lastRxInterface = nullptr; BaseSerialInterface* _lockedReplyInterface = nullptr; @@ -144,6 +145,7 @@ public: // enable all interfaces void enable() override { _enabled = true; + _nextRxIndex = 0; _lastRxInterface = nullptr; _lockedReplyInterface = nullptr; for(auto iface : _interfaces){ @@ -157,6 +159,7 @@ public: // disable all interfaces void disable() override { _enabled = false; + _nextRxIndex = 0; _lastRxInterface = nullptr; _lockedReplyInterface = nullptr; for(auto iface : _interfaces){ @@ -364,11 +367,15 @@ public: _lastRxInterface = nullptr; } - // try to read a frame from any enabled interface - for(auto iface : _interfaces){ + // Start after the last command so a pipelined client cannot starve the + // other transports. Locked response streams retain their exclusive poll. + for (size_t offset = 0; offset < MAX_INTERFACES; ++offset) { + const size_t index = (_nextRxIndex + offset) % MAX_INTERFACES; + auto iface = _interfaces[index]; if(iface.instance && iface.instance->isEnabled()){ size_t frameSize = iface.instance->checkRecvFrame(dest); if(frameSize > 0){ + _nextRxIndex = (index + 1) % MAX_INTERFACES; _lastRxInterface = iface.instance; return frameSize; } diff --git a/test/fixtures/companion_wifi_listener/test.cpp b/test/fixtures/companion_wifi_listener/test.cpp index 9a215c1c4..4c1584bd8 100644 --- a/test/fixtures/companion_wifi_listener/test.cpp +++ b/test/fixtures/companion_wifi_listener/test.cpp @@ -4,6 +4,8 @@ #include #include #include +#include +#include #include #include @@ -24,6 +26,31 @@ static Bytes framed(std::initializer_list frames) { return result; } +struct RoutingPeer : BaseSerialInterface { + bool enabled = false, connected = true; + unsigned polls = 0; + std::deque input; + std::vector output; + void enable() override { enabled = true; } + void disable() override { enabled = false; } + bool isEnabled() const override { return enabled; } + bool isConnected() const override { return connected; } + bool isReadBusy() const override { return !input.empty(); } + bool isWriteBusy() const override { return false; } + size_t writeFrame(const uint8_t* src, size_t length) override { + if (!enabled || !connected) return 0; + output.emplace_back(src, src + length); + return length; + } + size_t checkRecvFrame(uint8_t* dest) override { + ++polls; + if (!connected || input.empty()) return 0; + const Bytes frame = input.front(); input.pop_front(); + memcpy(dest, frame.data(), frame.size()); + return frame.size(); + } +}; + struct Fixture { SerialWifiInterface transport; uint8_t input[MAX_FRAME_SIZE] = {}; @@ -601,6 +628,152 @@ int main(int argc, char** argv) { f.transport.loop(); assert(!socket->connected && f.cancellations == 1); assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::FrameTimeout); + } else if (scenario == 52) { + assert(f.start()); + auto socket = f.connect(A); + socket->write_limit = std::numeric_limits::max(); + RoutingPeer usb; + MultiSerialInterface manager; + assert(manager.addInterface(InterfaceType::WiFi, &f.transport)); + assert(manager.addInterface(InterfaceType::USB, &usb)); manager.enable(); + const Bytes wifi_reply{0x0d, 0xa1}, usb_reply{0x0d, 0xb1}; + unsigned wifi_commands = 0; + for (unsigned poll = 0; poll < 10000; ++poll) { + const uint8_t query[] = {'<', 2, 0, 0x16, 3}; + for (uint8_t byte : query) socket->received.push_back(byte); + if (poll % 100 == 0) usb.input.push_back(Bytes{0x16, 4}); + ++mock_millis; manager.loop(); + assert(manager.checkRecvFrame(f.input) == 2 && f.input[0] == 0x16); + auto* route = manager.captureReplyRoute(); + assert(route == (f.input[1] == 3 ? static_cast(&f.transport) : &usb)); + const Bytes& reply = route == &usb ? usb_reply : wifi_reply; + if (route == &f.transport) ++wifi_commands; + assert(manager.writeFrame(reply.data(), reply.size()) == reply.size()); + assert((poll % 100 < 1 || usb.input.empty()) + && "an unlocked pipelined WiFi client must not starve a ready USB command"); + } + // Finish the finite WiFi pipeline and the last accepted response. + for (unsigned poll = 0; poll < 102; ++poll) { + const size_t length = manager.checkRecvFrame(f.input); + if (length == 0) break; + assert(length == 2 && manager.captureReplyRoute() == &f.transport); + ++wifi_commands; + assert(manager.writeFrame(wifi_reply.data(), wifi_reply.size()) == wifi_reply.size()); + } + assert(socket->received.empty() && wifi_commands == 10000); + assert(f.transport.getStatus().txFrames == wifi_commands); + Bytes expected; + const Bytes wire_reply = framed({wifi_reply}); + for (unsigned i = 0; i < wifi_commands; ++i) + expected.insert(expected.end(), wire_reply.begin(), wire_reply.end()); + assert(socket->sent == expected && usb.output == std::vector(100, usb_reply)); + } else if (scenario == 53) { + assert(f.start()); + auto socket = f.connect(A); + socket->write_limit = std::numeric_limits::max(); + RoutingPeer usb, ble, ethernet; + MultiSerialInterface manager; + BaseSerialInterface* routes[] = {&f.transport, &usb, &ble, ðernet}; + RoutingPeer* peers[] = {&usb, &ble, ðernet}; + const InterfaceType types[] = {InterfaceType::WiFi, InterfaceType::USB, + InterfaceType::Bluetooth, InterfaceType::Ethernet}; + for (unsigned i = 0; i < 4; ++i) assert(manager.addInterface(types[i], routes[i])); + manager.enable(); + socket->received = {'<', 2, 0, 0x16, 0}; + for (unsigned i = 0; i < 3; ++i) peers[i]->input.push_back(Bytes{0x16, uint8_t(i + 1)}); + unsigned counts[4] = {}; + BaseSerialInterface* captured_wifi = nullptr; + for (unsigned poll = 0; poll < 16000; ++poll) { + ++mock_millis; manager.loop(); + assert(manager.checkRecvFrame(f.input) == 2); + const unsigned index = poll % 4; + assert(manager.captureReplyRoute() == routes[index] + && "each unlocked ready transport must be served within one registration cycle"); + assert(f.input[0] == 0x16 && f.input[1] == index); + ++counts[index]; + const uint8_t reply[] = {0x0d, uint8_t(index)}; + assert(manager.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + if (index == 0) { + captured_wifi = manager.captureReplyRoute(); + if (counts[index] < 4000) socket->received = {'<', 2, 0, 0x16, 0}; + } else if (counts[index] < 4000) { + peers[index - 1]->input.push_back(Bytes{0x16, uint8_t(index)}); + } + } + assert(manager.checkRecvFrame(f.input) == 0); // Drain WiFi's final response. + for (unsigned count : counts) assert(count == 4000); + for (unsigned i = 0; i < 3; ++i) + assert(peers[i]->output == std::vector(4000, Bytes{0x0d, uint8_t(i + 1)})); + Bytes expected; + for (unsigned i = 0; i < 4000; ++i) { + const Bytes frame = framed({Bytes{0x0d, 0}}); + expected.insert(expected.end(), frame.begin(), frame.end()); + } + assert(socket->sent == expected && f.transport.getStatus().txFrames == 4000); + // A captured delayed route is independent of the most recent requester. + assert(manager.captureReplyRoute() == ðernet && captured_wifi == &f.transport); + const uint8_t delayed[] = {0x85, 0x41}; + assert(manager.writeFrameToRoute(captured_wifi, delayed, sizeof(delayed)) == sizeof(delayed)); + assert(manager.checkRecvFrame(f.input) == 0); + const Bytes last = framed({Bytes{0x85, 0x41}}); + expected.insert(expected.end(), last.begin(), last.end()); + assert(socket->sent == expected && ethernet.output.size() == 4000); + } else if (scenario == 54) { + assert(f.start()); + auto socket = f.connect(A, Bytes{'<', 2, 0, 0x16, 3}); + RoutingPeer usb, ble; + MultiSerialInterface manager; + assert(manager.addInterface(InterfaceType::WiFi, &f.transport)); + assert(manager.addInterface(InterfaceType::USB, &usb)); + assert(manager.addInterface(InterfaceType::Bluetooth, &ble)); manager.enable(); + socket->received = {'<', 2, 0, 0x16, 3}; + assert(manager.checkRecvFrame(f.input) == 2); manager.lockReplyRoute(); + usb.input.push_back(Bytes{0x16, 4}); ble.input.push_back(Bytes{0x16, 5}); + const uint8_t reply[] = {0x03, 0x41}; + for (unsigned poll = 0; poll < 8; ++poll) { + socket->received = {'<', 2, 0, 0x16, 3}; + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &f.transport); + assert(manager.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + socket->write_limit = std::numeric_limits::max(); + } + assert(manager.checkRecvFrame(f.input) == 0 && manager.captureReplyRoute() == &f.transport); + assert(usb.polls == 0 && ble.polls == 0 && usb.output.empty() && ble.output.empty()); + socket->connected = false; manager.loop(); + assert(!manager.isConnected() && manager.checkRecvFrame(f.input) == 0); + assert(manager.writeFrame(reply, sizeof(reply)) == 0); + assert(usb.polls == 0 && ble.polls == 0); + // The response producer owns cancellation; forgetting its disconnected + // route releases the lock without leaking the abandoned response. + manager.forgetReplyRouteForDisconnected(&f.transport); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &usb); + assert(manager.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + assert((usb.output == std::vector{Bytes{0x03, 0x41}} && ble.output.empty())); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &ble); + } else if (scenario == 55) { + RoutingPeer first, removed, third, fourth, replacement; + RoutingPeer* peers[] = {&first, &removed, &third, &fourth}; + MultiSerialInterface manager; + for (auto* peer : peers) assert(manager.addInterface(InterfaceType::USB, peer)); + manager.enable(); + for (auto* peer : peers) peer->input.push_back(Bytes{0x16, 3}); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &first); + assert(manager.removeInterface(&removed)); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &third); + assert(manager.addInterface(InterfaceType::Ethernet, &replacement)); replacement.enable(); + replacement.input.push_back(Bytes{0x16, 3}); first.input.push_back(Bytes{0x16, 3}); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &fourth); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &first); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &replacement); + third.input.push_back(Bytes{0x16, 3}); third.disable(); + fourth.input.push_back(Bytes{0x16, 3}); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &fourth); + assert(third.input.size() == 1 && removed.input.size() == 1); + first.input.push_back(Bytes{0x16, 3}); manager.disable(); + assert(manager.checkRecvFrame(f.input) == 0 && manager.captureReplyRoute() == nullptr); + assert(first.input.size() == 1); + manager.enable(); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &first); + assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &third); } else { assert(false && "unknown scenario"); } diff --git a/test/test_companion_wifi_listener.py b/test/test_companion_wifi_listener.py index 97d0b4ce3..d2407b185 100644 --- a/test/test_companion_wifi_listener.py +++ b/test/test_companion_wifi_listener.py @@ -73,10 +73,51 @@ SCENARIOS = ( "reply-capacity pause preserves the remaining inbound timeout budget", "new queue pressure cannot retroactively forgive an expired inbound frame", "reply-capacity timeout pause survives millis rollover", + "pipelined WiFi commands cannot starve a ready USB command or steal its replies", + "four continuously ready transports each receive bounded service and preserve captured replies", + "locked WiFi owner remains exclusive through idle and disconnect until route cleanup", + "round-robin skips removed and disabled slots and restarts registration order on enable", ) class CompanionWiFiListenerTests(unittest.TestCase): + def test_fixed_priority_negative_control_reproduces_cross_transport_starvation(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") + header = (ROOT / "src/helpers/MultiSerialInterface.h").read_text(encoding="ascii") + rotating_index = "const size_t index = (_nextRxIndex + offset) % MAX_INTERFACES;" + self.assertEqual(header.count(rotating_index), 1) + # Retain all routing, locking, and backend code. Only restore the old + # fixed-priority scan, which always started at registration slot zero. + legacy = header.replace(rotating_index, "const size_t index = offset;") + with tempfile.TemporaryDirectory(prefix="meshcore-wifi-priority-") as directory: + work = Path(directory) + (work / "helpers").mkdir() + (work / "helpers/MultiSerialInterface.h").write_text(legacy, encoding="ascii") + binary = work / "fixed-priority.exe" + compiled = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + *SANITIZERS, "-DESP32=1", f"-I{work}", f"-I{MOCKS}", + f"-I{ROOT / 'src'}", f"-I{ROOT / 'src/helpers'}", + str(FIXTURE), str(ROOT / "src/helpers/wifi/SerialWifiInterface.cpp"), + "-o", str(binary)], capture_output=True, text=True, timeout=60, + ) + self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr) + for scenario, oracle in ( + (52, "an unlocked pipelined WiFi client must not starve a ready USB command"), + (53, "each unlocked ready transport must be served within one registration cycle")): + with self.subTest(scenario=scenario): + checked = subprocess.run([str(binary), str(scenario)], capture_output=True, + text=True, timeout=10) + self.assertNotEqual(checked.returncode, 0) + self.assertIn(oracle, checked.stderr) + # Exclusive response streams are an intentional policy and were + # already correct; the old polling order must still pass this. + checked = subprocess.run([str(binary), "54"], capture_output=True, text=True, timeout=10) + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + self.assertIn("PASS: Wi-Fi listener scenario 54", checked.stdout) + def test_production_listener_faults(self): compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") if compiler is None: