diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 0683ad05..cdc67615 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -369,7 +369,8 @@ void MyMesh::writeDisabledFrame() { _serial->writeFrame(buf, 1); } -bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact) { +bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact, + BaseSerialInterface* route) { int i = 0; out_frame[i++] = code; memcpy(&out_frame[i], contact.id.pub_key, PUB_KEY_SIZE); @@ -389,7 +390,8 @@ bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact) { i += 4; memcpy(&out_frame[i], &contact.lastmod, 4); i += 4; - return _serial->writeFrame(out_frame, i) == (size_t)i; + return (route != NULL ? _serial->writeFrameToRoute(route, out_frame, i) + : _serial->writeFrame(out_frame, i)) == (size_t)i; } void MyMesh::stopContactsIterator() { @@ -399,7 +401,7 @@ void MyMesh::stopContactsIterator() { _iter_start_pending = false; _iter_contact_pending = false; _iter_next_frame_at = 0; - if (_serial != NULL) _serial->unlockReplyRoute(); + _iter_reply_route = NULL; } static constexpr int CONTACT_UPDATE_FRAME_MIN_LEN = @@ -2271,6 +2273,7 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe _serial(NULL), _mota_source_control(NULL), telemetry(MAX_PACKET_PAYLOAD - 4), _store(&store), _ui(ui), _iter(0) { _iter_started = false; + _iter_reply_route = NULL; _iter_start_pending = false; _iter_contact_pending = false; _iter_next_frame_at = 0; @@ -5181,8 +5184,8 @@ void MyMesh::serviceSigningSession() { } } -void MyMesh::cancelSerialResponseStream() { - stopContactsIterator(); +void MyMesh::cancelSerialResponseStream(BaseSerialInterface* route) { + if (route == NULL || _iter_reply_route == route) stopContactsIterator(); } void MyMesh::cancelSerialOperationsForRoute(BaseSerialInterface* route) { @@ -5268,7 +5271,7 @@ void MyMesh::handleCmdFrame(size_t len) { MESH_DEBUG_PRINTLN("App %s connected", app_name); BaseSerialInterface* app_route = _serial->captureReplyRoute(); - stopContactsIterator(); // stop any left-over ContactsIterator + cancelSerialResponseStream(app_route); cancelSerialOperationsForRoute(app_route); int i = 0; out_frame[i++] = RESP_CODE_SELF_INFO; @@ -5535,9 +5538,10 @@ void MyMesh::handleCmdFrame(size_t len) { _iter_filter_since = 0; } - // CONTACTS_START, every CONTACT, and END_OF_CONTACTS are one response - // transaction. Keep them on the transport which requested the list. - _serial->lockReplyRoute(); + // Retain this stream's destination independently of later commands. + // BLE's paced contact sync must not hold USB/WiFi command polling until + // the entire list finishes, or redirect their replies to the BLE app. + _iter_reply_route = _serial->captureReplyRoute(); _iter = startContactsIterator(); _iter_started = true; _iter_start_pending = true; @@ -5550,7 +5554,7 @@ void MyMesh::handleCmdFrame(size_t len) { uint8_t reply[5]; reply[0] = RESP_CODE_CONTACTS_START; memcpy(&reply[1], &_iter_total_count, 4); // total, NOT filtered count - if (_serial->writeFrame(reply, 5) == 5) { + if (_serial->writeFrameToRoute(_iter_reply_route, reply, 5) == 5) { _iter_start_pending = false; _iter_next_frame_at = futureMillis(CONTACT_STREAM_FRAME_INTERVAL_MS); } @@ -10123,6 +10127,9 @@ void MyMesh::checkCLIRescueCmd() { void MyMesh::checkSerialInterface() { size_t len = _serial->checkRecvFrame(cmd_frame); + if (_iter_started && !_serial->isReplyRouteAvailable(_iter_reply_route)) { + stopContactsIterator(); + } if (!_serial->isConnected()) { stopContactsIterator(); return; @@ -10130,11 +10137,12 @@ void MyMesh::checkSerialInterface() { if (len > 0) { handleCmdFrame(len); - } else if (_iter_started // check if our ContactsIterator is 'running' + } + if (_iter_started // check if our ContactsIterator is 'running' && (_iter_next_frame_at == 0 || _iter_next_frame_at == _ms->getMillis() || millisHasNowPassed(_iter_next_frame_at)) - && !_serial->isWriteBusy() // don't spam the Serial Interface too quickly! + && !_serial->isReplyRouteWriteBusy(_iter_reply_route) ) { if (_iter_table_revision != getContactTableRevision()) { // The paced stream walks the live compacting table. If an advert, app @@ -10154,7 +10162,8 @@ void MyMesh::checkSerialInterface() { uint8_t reply[5]; reply[0] = RESP_CODE_CONTACTS_START; memcpy(&reply[1], &_iter_total_count, 4); - if (_serial->writeFrame(reply, sizeof(reply)) == sizeof(reply)) { + if (_serial->writeFrameToRoute(_iter_reply_route, reply, sizeof(reply)) + == sizeof(reply)) { _iter_start_pending = false; _iter_next_frame_at = futureMillis(CONTACT_STREAM_FRAME_INTERVAL_MS); } @@ -10171,7 +10180,7 @@ void MyMesh::checkSerialInterface() { out_frame[0] = RESP_CODE_END_OF_CONTACTS; memcpy(&out_frame[1], &_most_recent_lastmod, 4); // include the most recent lastmod, so app can update their 'since' - if (_serial->writeFrame(out_frame, 5) == 5) { + if (_serial->writeFrameToRoute(_iter_reply_route, out_frame, 5) == 5) { stopContactsIterator(); } return; @@ -10181,7 +10190,8 @@ void MyMesh::checkSerialInterface() { // Do not advance past a contact until its complete frame has been accepted // by the selected transport. A transiently full/lost route must not turn a // successful CONTACTS_START/END transaction into a silently short list. - if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact)) { + if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact, + _iter_reply_route)) { if (_iter_pending_contact.lastmod > _most_recent_lastmod) { _most_recent_lastmod = _iter_pending_contact.lastmod; } @@ -10346,7 +10356,8 @@ bool MyMesh::canRecoverUsbLogging() const { && !_terminal_trace_pending #endif && !(_serial && _serial->hasPendingIO()) - && !(_iter_started && _serial && _serial->isConnected()); + && !(_iter_started && _serial + && _serial->isReplyRouteAvailable(_iter_reply_route)); } bool MyMesh::hasPendingWork() const { @@ -10370,7 +10381,8 @@ bool MyMesh::hasPendingWork() const { const bool contact_write_needs_polling = isContactWriteDue(); #endif return (_serial != NULL && _serial->hasPendingIO()) - || (_iter_started && _serial != NULL && _serial->isConnected()) + || (_iter_started && _serial != NULL + && _serial->isReplyRouteAvailable(_iter_reply_route)) || hasPendingOtaApply() || command_radio_apply_pending || _scheduled_reboot_at != 0 diff --git a/examples/companion_radio/MyMesh.h b/examples/companion_radio/MyMesh.h index 4b3c7a40..a8570ee2 100644 --- a/examples/companion_radio/MyMesh.h +++ b/examples/companion_radio/MyMesh.h @@ -188,7 +188,7 @@ public: void activateRadio(); bool isRadioReady() const { return _radio_available; } void startInterface(BaseSerialInterface &serial); - void cancelSerialResponseStream(); + void cancelSerialResponseStream(BaseSerialInterface* route = NULL); void cancelSerialOperationsForRoute(BaseSerialInterface* route); bool hasFiniteDelayedReplyForRoute(BaseSerialInterface* route) const; void resetUsbHostSessionInput(); @@ -481,7 +481,8 @@ private: BaseSerialInterface* route = nullptr); size_t writePendingSerialFrame(const uint8_t frame[], size_t len, uint32_t now); void writeDisabledFrame(); - bool writeContactRespFrame(uint8_t code, const ContactInfo &contact); + bool writeContactRespFrame(uint8_t code, const ContactInfo &contact, + BaseSerialInterface* route = NULL); void stopContactsIterator(); bool updateContactFromFrame(ContactInfo &contact, uint32_t& last_mod, const uint8_t *frame, int len); bool addToOfflineQueue(const uint8_t frame[], int len); @@ -650,6 +651,7 @@ private: AbstractUITask* _ui; ContactsIterator _iter; + BaseSerialInterface* _iter_reply_route; ContactInfo _iter_pending_contact; uint32_t _iter_filter_since; uint32_t _iter_next_frame_at; diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index c0ed9d69..42d7e65c 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -628,9 +628,7 @@ static void serviceCompanionPowerSaving(bool force = false) { #if defined(ETHERNET_ENABLED) static void cancelCompanionEthernetSession(void*) { - if (interface_manager.isReplyRouteFor(ðernet_interface)) { - the_mesh.cancelSerialResponseStream(); - } + the_mesh.cancelSerialResponseStream(ðernet_interface); the_mesh.cancelSerialOperationsForRoute(ðernet_interface); interface_manager.forgetReplyRouteForDisconnected(ðernet_interface); } @@ -777,12 +775,9 @@ static void cancelUsbSerialOperations() { // A cancelled reply must not enable diagnostics for the next USB host. usb_logging_reply_pending = false; #endif - // Contact enumeration uses the manager's pinned streaming route; delayed - // single replies capture their own route inside MyMesh. Cancel only USB's - // ownership so a simultaneous BLE/WiFi operation keeps running. - if (interface_manager.isReplyRouteFor(&usb_serial_interface)) { - the_mesh.cancelSerialResponseStream(); - } + // Contact streams and delayed replies capture their own destinations. + // Cancel only USB's ownership so simultaneous BLE/WiFi work keeps running. + the_mesh.cancelSerialResponseStream(&usb_serial_interface); the_mesh.cancelSerialOperationsForRoute(&usb_serial_interface); interface_manager.forgetReplyRouteForDisconnected(&usb_serial_interface); } @@ -2308,9 +2303,7 @@ void halt() { } static void cancelCompanionWiFiSession(void*) { - if (interface_manager.isReplyRouteFor(&wifi_interface)) { - the_mesh.cancelSerialResponseStream(); - } + the_mesh.cancelSerialResponseStream(&wifi_interface); the_mesh.cancelSerialOperationsForRoute(&wifi_interface); interface_manager.forgetReplyRouteForDisconnected(&wifi_interface); } @@ -2753,6 +2746,7 @@ static bool hasCompanionNonBluetoothClient() { static void disableCompanionBluetoothForCli() { companion_bluetooth_off_at = 0; + the_mesh.cancelSerialResponseStream(&bluetooth_interface); the_mesh.cancelSerialOperationsForRoute(&bluetooth_interface); interface_manager.disableBluetooth(); interface_manager.forgetReplyRouteForDisconnected(&bluetooth_interface); @@ -3422,9 +3416,7 @@ void setup() { // add usb interface #if defined(RP2040_PLATFORM) && defined(ENABLE_WIFI_INTERFACE) wifi_interface.setSessionChangedCallback([](void*) { - if (interface_manager.isReplyRouteFor(&wifi_interface)) { - the_mesh.cancelSerialResponseStream(); - } + the_mesh.cancelSerialResponseStream(&wifi_interface); the_mesh.cancelSerialOperationsForRoute(&wifi_interface); interface_manager.forgetReplyRouteForDisconnected(&wifi_interface); }, nullptr); diff --git a/test/test_companion_bluetooth_control.py b/test/test_companion_bluetooth_control.py index 0b5d5f6d..55b52f8f 100644 --- a/test/test_companion_bluetooth_control.py +++ b/test/test_companion_bluetooth_control.py @@ -51,6 +51,7 @@ struct { bluetooth_preference=durable_bluetooth_preference=enabled;return true; } bool isAnyNetworkTerminalMode(){return network_terminal;} + void cancelSerialResponseStream(BaseSerialInterface*) {} void cancelSerialOperationsForRoute(BaseSerialInterface* route){ assert(route==&bluetooth_interface);++cancelled; } diff --git a/test/test_companion_contact_stream_concurrency.py b/test/test_companion_contact_stream_concurrency.py new file mode 100644 index 00000000..9cc4d61e --- /dev/null +++ b/test/test_companion_contact_stream_concurrency.py @@ -0,0 +1,280 @@ +#!/usr/bin/env python3 +"""Run production contact streaming with interleaved USB and paced BLE clients.""" +from pathlib import Path +import os +import re +import shutil +import subprocess +import sys +import tempfile +import unittest + +from test_replay_reset_integration import extract_braced + +ROOT = Path(__file__).resolve().parents[1] + +HARNESS = r''' +#include +#include +#include +#include +#include +#include +#include +#include +#include "helpers/ArduinoSerialInterface.h" +#include "helpers/MultiSerialInterface.h" +#define PUB_KEY_SIZE 32 +#define MAX_PATH_SIZE 64 +template static void ignoreDebug(Args&&...) {} +#define MESH_DEBUG_PRINTLN(...) ignoreDebug(__VA_ARGS__) +namespace StrHelper { +static void strzcpy(void* out,const char* in,size_t size) { + memset(out,0,size);memcpy(out,in,std::min(size,strlen(in))); +} +} +@CONSTANTS@ +@BUFFER@ +struct ContactInfo { + struct { uint8_t pub_key[32]{}; } id; + uint8_t type=1, flags=0, out_path_len=0; + char name[32]{}; + uint32_t last_advert_timestamp=0, gps_lat=0, gps_lon=0, lastmod=1; + bool copyPathTo(uint8_t* out) const { memset(out,0,MAX_PATH_SIZE); return true; } +}; +struct ContactsIterator { + unsigned next=0, count=0; + template bool hasNext(T*,ContactInfo& c) { + if(next==count) return false; + c=ContactInfo(); c.id.pub_key[0]=static_cast(++next); + snprintf(c.name,sizeof(c.name),"HIL-%u",next); + c.lastmod=next; return true; + } +}; +struct Ble : BaseSerialInterface { + bool enabled=false, connected=true; + unsigned last_write=0, capacity=4; + std::deque> input, pending; + std::vector> delivered; + 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 millis()-last_write<60; } + size_t writeFrame(const uint8_t* src,size_t n) override { + if(!enabled||!connected||pending.size()>=capacity) return 0; + pending.emplace_back(src,src+n); return n; + } + size_t checkRecvFrame(uint8_t* dest) override { + if(connected&&!pending.empty()&&!isWriteBusy()) { + delivered.push_back(pending.front()); pending.pop_front(); last_write=millis(); + } + if(!connected||input.empty()) return 0; + auto frame=input.front(); input.pop_front(); + memcpy(dest,frame.data(),frame.size()); return frame.size(); + } +}; +struct MyMesh { + BaseSerialInterface* _serial=nullptr; + struct Clock { unsigned getMillis() const {return millis();} } clock, *_ms=&clock; + struct Store { bool hasIncompleteContactLoad() const {return false;} } store, *_store=&store; + ContactsIterator _iter; + BaseSerialInterface* _iter_reply_route=nullptr; + ContactInfo _iter_pending_contact; + uint32_t _iter_filter_since=0, _iter_next_frame_at=0, _iter_total_count=0; + uint32_t _iter_table_revision=0, _most_recent_lastmod=0; + bool _iter_started=false, _iter_start_pending=false, _iter_contact_pending=false; + unsigned contacts=54, revision=1; + uint8_t out_frame[MAX_FRAME_SIZE+4]{},cmd_frame[MAX_FRAME_SIZE+1]{}; + bool millisHasNowPassed(uint32_t time) const {return static_cast(time-millis())<0;} + uint32_t futureMillis(uint32_t delta) const {return millis()+delta;} + uint32_t getContactTableRevision() const {return revision;} + unsigned getNumContacts() const {return contacts;} + ContactsIterator startContactsIterator() {return {0,contacts};} + void writeErrFrame(uint8_t error) { + uint8_t reply[]={RESP_CODE_ERR,error}; assert(_serial->writeFrame(reply,2)==2); + } + void cancelSerialOperationsForRoute(BaseSerialInterface*) {} + void stopContactsIterator(); + void cancelSerialResponseStream(BaseSerialInterface* route=nullptr); + bool writeContactRespFrame(uint8_t,const ContactInfo&,BaseSerialInterface* route=nullptr); + void checkSerialInterface(); + void handleCmdFrame(size_t len) { + @GET_CONTACTS@ + else if(cmd_frame[0]==CMD_APP_START&&len>=8) { + @APP_BEGIN@ + memset(out_frame,0xA5,70);out_frame[0]=RESP_CODE_SELF_INFO; + assert(_serial->writeFrame(out_frame,70)==70); + } else { + assert(cmd_frame[0]==CMD_GET_BATT_AND_STORAGE); + memset(out_frame,0x5A,11);out_frame[0]=RESP_CODE_BATT_AND_STORAGE; + assert(_serial->writeFrame(out_frame,11)==11); + } + } +}; +@METHODS@ +struct Fixture { + BufferStream stream; + ArduinoSerialInterface usb; + Ble ble; + MultiSerialInterface manager; + MyMesh mesh; + bool drain_usb=true; + Fixture() { + g_mock_millis=100; + usb.begin(stream); usb.enableFlowControl(true); + assert(manager.addInterface(InterfaceType::Bluetooth,&ble)); + assert(manager.addInterface(InterfaceType::USB,&usb)); manager.enable(); + mesh._serial=&manager; + } + void step() { + if(drain_usb) stream.write_capacity=4096; + mesh.checkSerialInterface(); manager.loop(); ++g_mock_millis; + } + void run(unsigned count) {while(count--) step();} + void sendUsb(std::initializer_list payload) { + const uint8_t header[]={'<',static_cast(payload.size()),0}; + stream.push(header,3); stream.push(payload.begin(),payload.size()); + } + std::vector> usbFrames() const { + std::vector> frames; + for(size_t pos=0;pos'&&pos+3<=stream.output.size()); + size_t n=stream.output[pos+1]|(stream.output[pos+2]<<8); + assert(n>0&&pos+n+3<=stream.output.size()); + frames.emplace_back(stream.output.begin()+pos+3,stream.output.begin()+pos+3+n); + pos+=n+3; + } + return frames; + } + void startBle() {ble.input.push_back({CMD_GET_CONTACTS});step();assert(mesh._iter_started);} + void finish() {run(6000);assert(!mesh._iter_started);} +}; +static unsigned count(const std::vector>& frames,uint8_t type) { + return std::count_if(frames.begin(),frames.end(),[type](const auto& f){return f[0]==type;}); +} +static void assertBleSnapshot(const Fixture& f,unsigned contacts=54) { + assert(count(f.ble.delivered,RESP_CODE_CONTACTS_START)==1); + assert(count(f.ble.delivered,RESP_CODE_CONTACT)==contacts); + assert(count(f.ble.delivered,RESP_CODE_END_OF_CONTACTS)==1); + for(unsigned i=1;i<=contacts;++i) { + auto at=std::find_if(f.ble.delivered.begin(),f.ble.delivered.end(),[i](const auto& v){return v[0]==RESP_CODE_CONTACT&&v[1]==i;}); + assert(at!=f.ble.delivered.end()); + } +} +int main() { + // Fifty-four BLE contacts need over three seconds, but each independent USB + // request is handled immediately, even when it arrives on every loop pass. + { + Fixture f;f.startBle();unsigned queries=0,apps=0; + for(unsigned now=0;now<4000;++now) { + const bool app=now%97==96; + if(app) { + f.mesh.cancelSerialResponseStream(&f.usb); + f.manager.forgetReplyRouteForDisconnected(&f.usb);f.usb.resetSessionState(); + f.sendUsb({CMD_APP_START,0,0,0,0,0,0,0});++apps; + } else {f.sendUsb({CMD_GET_BATT_AND_STORAGE});++queries;} + f.step();assert(f.usbFrames().size()==queries+apps); + } + f.finish();assertBleSnapshot(f); + assert(count(f.usbFrames(),RESP_CODE_CONTACT)==0); + assert(count(f.usbFrames(),RESP_CODE_CONTACTS_START)==0); + assert(count(f.usbFrames(),RESP_CODE_SELF_INFO)==apps); + assert(count(f.usbFrames(),RESP_CODE_BATT_AND_STORAGE)==queries); + for(const auto& response:f.usbFrames()) { + assert(response.size()==(response[0]==RESP_CODE_SELF_INFO?70u:11u)); + assert(response.back()==(response[0]==RESP_CODE_SELF_INFO?0xA5:0x5A)); + } + } + // A second client cannot replace the sole contact iterator; its ERR remains + // on that client's route, while the BLE list keeps its original destination. + { + Fixture f;f.startBle();f.sendUsb({CMD_GET_CONTACTS,0xFF,0xFF,0xFF,0xFF});f.step(); + assert(f.usbFrames().size()==1&&f.usbFrames()[0][0]==RESP_CODE_ERR); + assert(f.usbFrames()[0][1]==ERR_CODE_BAD_STATE); + f.finish();assertBleSnapshot(f); + } + // Session loss cancels only its contact stream, irrespective of whichever + // interface supplied the newest command. No remaining contacts leak to USB. + { + Fixture f;f.startBle();f.sendUsb({CMD_GET_BATT_AND_STORAGE});f.step(); + f.mesh.cancelSerialResponseStream(&f.usb);assert(f.mesh._iter_started); + f.mesh.cancelSerialResponseStream(&f.ble);assert(!f.mesh._iter_started); + f.run(300);assert(count(f.usbFrames(),RESP_CODE_CONTACT)==0); + } + { + Fixture f;f.startBle();f.ble.connected=false;f.step();assert(!f.mesh._iter_started); + f.sendUsb({CMD_GET_BATT_AND_STORAGE});f.step();assert(f.usbFrames()[0][0]==RESP_CODE_BATT_AND_STORAGE); + } + // The cached contact survives a failed BLE admission and an intervening + // command which overwrites both shared command/output scratch buffers. + { + Fixture f;f.mesh.contacts=3;f.startBle();f.ble.capacity=0;f.run(70); + assert(f.mesh._iter_contact_pending&&f.mesh._iter_pending_contact.id.pub_key[0]==1); + f.sendUsb({CMD_GET_BATT_AND_STORAGE});f.step(); + assert(f.usbFrames()[0].back()==0x5A&&f.mesh._iter_contact_pending); + f.ble.capacity=4;f.finish();assertBleSnapshot(f,3); + } + { + Fixture f;f.startBle();f.ble.input.push_back({CMD_APP_START,0,0,0,0,0,0,0}); + f.step();assert(!f.mesh._iter_started);f.run(300); + assert(count(f.ble.delivered,RESP_CODE_SELF_INFO)==1); + assert(count(f.usbFrames(),RESP_CODE_SELF_INFO)==0); + } + // A backed-up USB contact stream also leaves BLE queries responsive; once + // capacity returns, every START/contact/END is retained on the USB route. + { + Fixture f;f.drain_usb=false;f.stream.write_capacity=0; + f.sendUsb({CMD_GET_CONTACTS});f.step();assert(f.mesh._iter_started); + f.mesh.cancelSerialResponseStream(&f.ble);assert(f.mesh._iter_started); + f.ble.input.push_back({CMD_GET_CONTACTS});f.step(); + f.ble.input.push_back({CMD_GET_BATT_AND_STORAGE});f.step();f.run(100); + assert(count(f.ble.delivered,RESP_CODE_ERR)==1); + assert(count(f.ble.delivered,RESP_CODE_BATT_AND_STORAGE)==1); + f.drain_usb=true;f.finish(); + assert(count(f.usbFrames(),RESP_CODE_CONTACTS_START)==1); + assert(count(f.usbFrames(),RESP_CODE_CONTACT)==54); + assert(count(f.usbFrames(),RESP_CODE_END_OF_CONTACTS)==1); + assert(count(f.ble.delivered,RESP_CODE_CONTACT)==0); + } +} +''' + + +class ContactStreamConcurrencyTests(unittest.TestCase): + def test_paced_stream_keeps_routes_without_blocking_other_clients(self): + compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++') + if not compiler: + self.skipTest('a host C++17 compiler is required') + mesh = (ROOT / 'examples/companion_radio/MyMesh.cpp').read_text() + native = (ROOT / 'test/test_serial_mode_switch/test_serial_mode_switch.cpp').read_text() + constants = '\n'.join(re.findall( + r'^#define (?:CONTACT_STREAM_FRAME_INTERVAL_MS|CMD_GET_CONTACTS|CMD_APP_START|CMD_GET_BATT_AND_STORAGE|RESP_CODE_ERR|RESP_CODE_CONTACTS_START|RESP_CODE_CONTACT|RESP_CODE_END_OF_CONTACTS|RESP_CODE_SELF_INFO|RESP_CODE_BATT_AND_STORAGE|ERR_CODE_BAD_STATE|ERR_CODE_FILE_IO_ERROR)\s+.+$', + mesh, re.MULTILINE)) + contacts = extract_braced(mesh, '} else if (cmd_frame[0] == CMD_GET_CONTACTS').replace('} else if', 'if', 1) + app = extract_braced(mesh, '} else if (cmd_frame[0] == CMD_APP_START') + app = app[app.index('{') + 1:app.index(' int i = 0;')] + methods = '\n'.join(extract_braced(mesh, sig) for sig in ( + 'bool MyMesh::writeContactRespFrame(', 'void MyMesh::stopContactsIterator(', + 'void MyMesh::cancelSerialResponseStream(', 'void MyMesh::checkSerialInterface()', + )) + program = HARNESS.replace('@CONSTANTS@', constants).replace('@BUFFER@', extract_braced(native, 'class BufferStream') + ';') + program = program.replace('@GET_CONTACTS@', contacts).replace('@APP_BEGIN@', app).replace('@METHODS@', methods) + sanitizers = ['-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie'] if sys.platform.startswith('linux') else [] + with tempfile.TemporaryDirectory(prefix='meshcore-contact-concurrency-') as directory: + binary = Path(directory) / 'concurrency' + result = subprocess.run([ + compiler, '-std=c++17', '-Wall', '-Wextra', '-Werror', + '-Wno-unused-parameter', '-Wno-sign-compare', + *sanitizers, + f'-I{ROOT / "test/mocks"}', f'-I{ROOT / "src"}', '-x', 'c++', '-', + str(ROOT / 'src/helpers/ArduinoSerialInterface.cpp'), '-o', str(binary), + ], input=program, text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(binary)], text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_companion_contact_stream_contract.py b/test/test_companion_contact_stream_contract.py index e6c243bb..3854f06f 100644 --- a/test/test_companion_contact_stream_contract.py +++ b/test/test_companion_contact_stream_contract.py @@ -30,18 +30,16 @@ class ContactStreamContractTests(unittest.TestCase): def test_contact_frame_write_result_is_observable(self): header = source("examples/companion_radio/MyMesh.h") impl = source("examples/companion_radio/MyMesh.cpp") - self.assertIn( - "bool writeContactRespFrame(uint8_t code, const ContactInfo &contact);", - header, - ) + self.assertRegex(header, r"bool writeContactRespFrame\(uint8_t code,\s*" + r"const ContactInfo &contact,\s*" + r"BaseSerialInterface\* route = NULL\);") body = function_body( impl, - "bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact)", - ) - self.assertRegex( - body, - r"return\s+_serial->writeFrame\(out_frame,\s*i\)\s*==\s*\(size_t\)i\s*;", + "bool MyMesh::writeContactRespFrame(", ) + self.assertIn("_serial->writeFrameToRoute(route, out_frame, i)", body) + self.assertIn("_serial->writeFrame(out_frame, i)", body) + self.assertRegex(body, r"\)\s*==\s*\(size_t\)i\s*;") def test_iterator_retries_start_contact_and_end_frames(self): impl = source("examples/companion_radio/MyMesh.cpp") @@ -52,17 +50,18 @@ class ContactStreamContractTests(unittest.TestCase): self.assertRegex( body, r"if\s*\(writeContactRespFrame\(RESP_CODE_CONTACT,\s*" - r"_iter_pending_contact\)\)", + r"_iter_pending_contact,\s*_iter_reply_route\)\)", ) self.assertRegex( body, - r"if\s*\(_serial->writeFrame\(out_frame,\s*5\)\s*==\s*5\)\s*\{\s*" + r"if\s*\(_serial->writeFrameToRoute\(_iter_reply_route,\s*" + r"out_frame,\s*5\)\s*==\s*5\)\s*\{\s*" r"stopContactsIterator\(\)\s*;", ) # The cached contact is cleared only inside the successful-write arm. success = body.index( - "if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact))" + "if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact," ) clear = body.index("_iter_contact_pending = false;", success) self.assertGreater(clear, success) diff --git a/test/test_companion_radio_settings_transaction.py b/test/test_companion_radio_settings_transaction.py index 3c4e27af..e2dc03b5 100644 --- a/test/test_companion_radio_settings_transaction.py +++ b/test/test_companion_radio_settings_transaction.py @@ -238,6 +238,9 @@ static uint32_t now_ms=100; struct Serial { bool hasPendingIO() const { return false; } bool isConnected() const { return false; } + bool isReplyRouteAvailable(const Serial* route) const { + return route==this&&isConnected(); + } }; struct { bool isWatchdogObserving() const { return false; } @@ -253,6 +256,7 @@ struct MyMesh { const void *sign_data=nullptr; const void *emergency_client_repeat_packet=nullptr; Serial* _serial=nullptr; + Serial* _iter_reply_route=nullptr; #if COMPANION_FEATURE_TEMP_RADIO bool _temp_radio_applied=false; uint32_t _temp_radio_set_at=0, _temp_radio_revert_at=0, _temp_radio_retry_at=0; diff --git a/test/test_companion_usb_logging_reply.py b/test/test_companion_usb_logging_reply.py index e8c15c7b..ae5a203c 100644 --- a/test/test_companion_usb_logging_reply.py +++ b/test/test_companion_usb_logging_reply.py @@ -96,7 +96,9 @@ struct MyMesh { uint8_t cmd_frame[MAX_FRAME_SIZE+1]{}, out_frame[MAX_FRAME_SIZE+4]{}; char reply_buf[166]{}; bool isTerminalMode() const { return terminal; } - void cancelSerialResponseStream() { ++cancellations; } + void cancelSerialResponseStream(BaseSerialInterface* route) { + if (interface_manager.isReplyRouteFor(route)) ++cancellations; + } void cancelSerialOperationsForRoute(BaseSerialInterface*) {} void enterTerminalMode(bool show_banner) { terminal=true; diff --git a/test/test_companion_wifi_session.py b/test/test_companion_wifi_session.py index fb1de723..729e8042 100644 --- a/test/test_companion_wifi_session.py +++ b/test/test_companion_wifi_session.py @@ -58,7 +58,8 @@ struct MyMesh { uint8_t private_key_backup_sender[6] = {}; struct Ack { BaseSerialInterface* reply_route; bool radio_retry; }; Ack expected_ack_table[EXPECTED_ACK_TABLE_SIZE] = {}; - void cancelSerialResponseStream() { + void cancelSerialResponseStream(BaseSerialInterface* route) { + if (!interface_manager.isReplyRouteFor(route)) return; ++stream_cancels; if (streaming) { streaming = false; interface_manager.unlockReplyRoute(); } } diff --git a/test/test_nrf52_usb_logging_contract.py b/test/test_nrf52_usb_logging_contract.py index acf431cd..ad189660 100644 --- a/test/test_nrf52_usb_logging_contract.py +++ b/test/test_nrf52_usb_logging_contract.py @@ -520,11 +520,8 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase): main.index("static void cancelUsbSerialOperations()"): main.index("static void enterUsbTerminalMode(") ] - route_check = usb_cancel.index( - "interface_manager.isReplyRouteFor(&usb_serial_interface)" - ) route_cancel = usb_cancel.index( - "the_mesh.cancelSerialResponseStream()" + "the_mesh.cancelSerialResponseStream(&usb_serial_interface)" ) delayed_cancel = usb_cancel.index( "the_mesh.cancelSerialOperationsForRoute(&usb_serial_interface)" @@ -532,7 +529,6 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase): route_forget = usb_cancel.index( "interface_manager.forgetReplyRouteForDisconnected(" ) - self.assertLess(route_check, route_cancel) self.assertLess(route_cancel, delayed_cancel) self.assertLess(delayed_cancel, route_forget) self.assertIn("the_mesh.resetUsbHostSessionInput();", reset_helper) diff --git a/test/test_usb_logging_watchdog_roles.py b/test/test_usb_logging_watchdog_roles.py index 3ccb9178..3b63db12 100644 --- a/test/test_usb_logging_watchdog_roles.py +++ b/test/test_usb_logging_watchdog_roles.py @@ -39,6 +39,9 @@ struct SerialInterface { bool pending=false, connected=false, passthrough=true; bool hasPendingIO() const{return pending;} bool isConnected() const{return connected;} + bool isReplyRouteAvailable(const SerialInterface* route) const { + return route==this&&connected; + } bool isPassthroughMode() const{return passthrough;} }; struct RoleMesh { @@ -54,6 +57,7 @@ struct RoleMesh { Store* _store=&store; SerialInterface serial; SerialInterface* _serial=&serial; + SerialInterface* _iter_reply_route=&serial; bool hasOutbound() const{return outbound;} bool isAnyTempRadioActive() const{return temporary;} bool hasPendingOtaApply() const{return ota_apply;} @@ -84,6 +88,7 @@ void reset(){ // Assignment copies the fixture's dependency pointers; point them back at // the live fixture objects, not the temporary used to reset scalar state. the_mesh._store=&the_mesh.store;the_mesh._serial=&the_mesh.serial; + the_mesh._iter_reply_route=&the_mesh.serial; command[0]=0;command_overflow=false; #if defined(ENABLE_USB_INTERFACE) usb_serial_interface=SerialInterface{};usb_connected=false;