From 5e5e608dad259eb84bfb4784f48d256a5bae5ecd Mon Sep 17 00:00:00 2001 From: flower_linux <737666956@qq.com> Date: Tue, 19 May 2026 16:38:09 +0800 Subject: [PATCH] first commit --- CMakeLists.txt | 28 +- CMakeLists.txt.user | 32 +- README.md | 44 +- config/edge_manifest.example.json | 15 +- config/edge_manifest.schema.json | 52 ++- src/device_bus/EdgeDeviceBus.cpp | 34 +- src/device_bus/EdgeDeviceBus.h | 9 + .../manager/SerialDeviceManager.cpp | 37 +- src/device_bus/manager/SerialDeviceManager.h | 7 + .../monitor/EdgeDiscoveryCoordinator.cpp | 115 ++++- .../monitor/EdgeDiscoveryCoordinator.h | 20 +- src/devices/physical/SerialDevice.cpp | 42 +- src/devices/physical/SerialDevice.h | 4 + src/fake_driver.cpp | 20 - src/fake_driver.hpp | 22 - .../drivers/serial/SerialQtDriver.cpp | 66 ++- src/hardware/drivers/serial/SerialQtDriver.h | 19 +- src/main.cpp | 381 ++++++++++++++--- src/message_bus/ingress/EdgeSession.cpp | 201 ++++++++- src/message_bus/ingress/EdgeSession.h | 52 +++ src/message_bus/ingress/EdgeTcpUplinkPort.cpp | 15 +- src/message_bus/ingress/EdgeUplinkWire.h | 2 + src/ops_ui_http.cpp | 58 ++- src/ops_ui_http.hpp | 7 +- src/pipeline/EdgeStreamPipeline.cpp | 399 ++++++++++++++++++ src/pipeline/EdgeStreamPipeline.h | 119 ++++++ src/utils/AgentDebugNdjson5448.h | 35 ++ 27 files changed, 1643 insertions(+), 192 deletions(-) delete mode 100644 src/fake_driver.cpp delete mode 100644 src/fake_driver.hpp create mode 100644 src/pipeline/EdgeStreamPipeline.cpp create mode 100644 src/pipeline/EdgeStreamPipeline.h create mode 100644 src/utils/AgentDebugNdjson5448.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 2eb93ec..5f68d4c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -20,8 +20,8 @@ set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin) set(EDGE_AGENT_SOURCES src/main.cpp - src/fake_driver.cpp - src/fake_driver.hpp + src/pipeline/EdgeStreamPipeline.cpp + src/pipeline/EdgeStreamPipeline.h src/device_bus/monitor/DeviceEvent.h src/device_bus/monitor/IDeviceMonitor.h src/device_bus/monitor/UdevDeviceMonitor.cpp @@ -62,18 +62,42 @@ endif() qt_add_executable(softbus_edge_agent ${EDGE_AGENT_SOURCES}) +set(SOFTBUS_DAEMON_ROOT "${CMAKE_CURRENT_SOURCE_DIR}/../softbus_daemon") +target_sources(softbus_edge_agent PRIVATE + ${SOFTBUS_DAEMON_ROOT}/src/core/plugin_system/PluginHost.cpp + ${SOFTBUS_DAEMON_ROOT}/src/core/plugin_system/PluginManager.cpp + ${SOFTBUS_DAEMON_ROOT}/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp +) + target_include_directories(softbus_edge_agent PRIVATE ${CMAKE_SOURCE_DIR}/src + ${SOFTBUS_DAEMON_ROOT}/src + ${SOFTBUS_DAEMON_ROOT} ) target_link_libraries(softbus_edge_agent PRIVATE Qt6::Core Qt6::Network Qt6::SerialPort) +if(UNIX AND NOT APPLE) + target_link_libraries(softbus_edge_agent PRIVATE dl) +endif() if(SOFTBUS_HAVE_CAN) target_link_libraries(softbus_edge_agent PRIVATE Qt6::SerialBus) endif() target_compile_definitions(softbus_edge_agent PRIVATE SOFTBUS_HAVE_CAN=${SOFTBUS_HAVE_CAN}) +# 与 daemon 侧 ASan 编译的协议 .so 对齐：Debug/RelWithDebInfo 下主程序也链 ASan，否则 dlopen 会报 +# "ASan runtime does not come first" 并退出。Release 不受影响。可用 -DSOFTBUS_EDGE_AGENT_DEBUG_ASAN=OFF 关闭。 +option(SOFTBUS_EDGE_AGENT_DEBUG_ASAN "Link AddressSanitizer in Debug/RelWithDebInfo (non-Apple Unix)" ON) +if(SOFTBUS_EDGE_AGENT_DEBUG_ASAN AND UNIX AND NOT APPLE) + if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") + target_compile_options(softbus_edge_agent PRIVATE + $<$,$>:-fsanitize=address>) + target_link_options(softbus_edge_agent PRIVATE + $<$,$>:-fsanitize=address>) + endif() +endif() + if(UNIX AND NOT APPLE) find_package(PkgConfig QUIET) if(PkgConfig_FOUND) diff --git a/CMakeLists.txt.user b/CMakeLists.txt.user index 4b3ba3c..41fa1bc 100644 --- a/CMakeLists.txt.user +++ b/CMakeLists.txt.user @@ -1,6 +1,6 @@ - + EnvironmentId @@ -93,29 +93,29 @@ Desktop true - Qt 6.9.2 - Qt 6.9.2 - {ac766883-1e6c-4832-9072-86a8b976b89e} + Desktop Qt 6.9.2 + Desktop Qt 6.9.2 + qt.qt6.692.linux_gcc_64_kit 0 0 0 - Build + Debug 2 false - -DCMAKE_PREFIX_PATH:PATH=%{Qt:QT_INSTALL_PREFIX} --DCMAKE_BUILD_TYPE:STRING=Build + -DCMAKE_GENERATOR:STRING=Ninja -DCMAKE_CXX_COMPILER:FILEPATH=%{Compiler:Executable:Cxx} --DCMAKE_GENERATOR:STRING=Unix Makefiles --DCMAKE_CXX_FLAGS_INIT:STRING=%{Qt:QML_DEBUG_FLAG} --DQT_MAINTENANCE_TOOL:FILEPATH=/home/dt/app/qt/MaintenanceTool --DQT_QMAKE_EXECUTABLE:FILEPATH=%{Qt:qmakeExecutable} --DCMAKE_PROJECT_INCLUDE_BEFORE:FILEPATH=%{BuildConfig:BuildDirectory:NativeFilePath}/.qtc/package-manager/auto-setup.cmake -DCMAKE_C_COMPILER:FILEPATH=%{Compiler:Executable:C} --DCMAKE_COLOR_DIAGNOSTICS:BOOL=ON - /home/dt/myCode/bus/soft_bus/edge_agent - /home/dt/myCode/bus/soft_bus/edge_agent/build +-DCMAKE_COLOR_DIAGNOSTICS:BOOL=ON +-DCMAKE_CXX_FLAGS_INIT:STRING=%{Qt:QML_DEBUG_FLAG} +-DCMAKE_PREFIX_PATH:PATH=%{Qt:QT_INSTALL_PREFIX} +-DCMAKE_PROJECT_INCLUDE_BEFORE:FILEPATH=%{BuildConfig:BuildDirectory:NativeFilePath}/.qtc/package-manager/auto-setup.cmake +-DQT_QMAKE_EXECUTABLE:FILEPATH=%{Qt:qmakeExecutable} +-DCMAKE_BUILD_TYPE:STRING=Debug +-DQT_MAINTENANCE_TOOL:FILEPATH=/home/dt/app/qt/MaintenanceTool + 0 + /home/dt/myCode/bus/soft_bus/edge_agent/build/Desktop_Qt_6_9_2-Debug @@ -155,7 +155,7 @@ false - 构建 (imported) + Debug CMakeProjectManager.CMakeBuildConfiguration 0 0 diff --git a/README.md b/README.md index cafe21b..8e25e66 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,14 @@ # softbus_edge_agent -独立边缘进程：连接核心 `softbus_daemon` 的 **edgeIngress** TCP 端口，发送 **HELLO**、**DATA_RAWBUS**，以及通过极简 **本地 HTTP**（`GET /`、`GET /ui`、`POST /propose` 等）触发 **TREE_PROPOSE** 反向同步。 +独立边缘进程：连接核心 `softbus_daemon` 的 **edgeIngress** TCP 端口，发送 **HELLO**、**DEBUG_TUNNEL**（含周期性边缘心跳）、**DATA_RAWBUS**（真实总线数据），以及通过极简 **本地 HTTP**（`GET /`、`GET /ui`、`POST /propose` 等）触发 **TREE_PROPOSE** 反向同步。 ## 目录布局（与 `softbus_daemon` 的 `src/` 对齐） | 路径 | 说明 | |------|------| | `src/device_bus/tree/` | 与核心同源的 `DeviceTreeModel`（本地子树、`TREE_FULL` 落盘加载） | -| `src/device_bus/monitor/` | **udev 串口发现**、`EdgeDiscoveryCoordinator`（防抖后组 `TREE_PROPOSE` / `upsert` 向核心注册） | -| `src/device_bus/manager/` | `SerialDeviceManager`、`CanDeviceManager`（无 Registry / Pipeline，仅 `IIngressPort` + 内存池） | +| `src/device_bus/monitor/` | **udev 串口发现**、`EdgeDiscoveryCoordinator`（防抖后：有核心则 `TREE_PROPOSE`，无核心则写本地树并 reconcile） | +| `src/device_bus/manager/` | `SerialDeviceManager`、`CanDeviceManager`；串口可选 **`EdgeStreamPipeline`**（daemon 同源 `PluginHost` + framer/parser/mapper） | | `src/device_bus/EdgeDeviceBus.*` | 边缘总线编排：加载本地树、**reconcile** 串口与 **SocketCAN** 实例 | | `src/devices/`、`src/hardware/` | 与核心同源的 `SerialDevice`、`SerialQtDriver` 等 | | `src/message_bus/ingress/` | `EdgeSession`、`EdgeUplinkWire`、`EdgeTcpUplinkPort`（`RawBusMessage` → SBUP 上行） | @@ -22,6 +22,9 @@ ```bash cmake -S . -B build -DCMAKE_PREFIX_PATH=/path/to/Qt/6.x/gcc_64 cmake --build build +# 无参：使用 $HOME/softbus/edge/edge_manifest.json（首次启动会自动创建目录并生成默认 manifest 与 local_tree.json） +./build/bin/softbus_edge_agent +# 或显式指定 manifest（例如仓库内示例；示例中 localTreePath 使用 $HOME/ 前缀便于各用户通用）： ./build/bin/softbus_edge_agent config/edge_manifest.example.json ``` @@ -29,7 +32,7 @@ cmake --build build ## 配置 `edge_manifest.json`（可选） -若默认路径 **`config/edge_manifest.json`** 不存在、或传入的 manifest 路径无法打开、或 JSON 非法，进程会使用**内置默认值**启动（等价于示例 manifest 的字段），并在日志中提示。 +无命令行参数时，默认 manifest 为 **`$HOME/softbus/edge/edge_manifest.json`**（启动前会 `mkpath` 该目录；若缺少 `edge_manifest.json` / `local_tree.json` 则写入首次启动模板）。若该文件仍无法打开、或 JSON 非法、或你传入的其它 manifest 路径无效，进程会使用**内置默认值**合并/回退，并在日志中提示。 仍可通过命令行指定路径覆盖默认文件位置： @@ -41,16 +44,26 @@ cmake --build build 要点： -- **`coreHost` / `corePort`**：与 `daemon_config.json` 中 `edgeIngress` 一致。 +- **`coreHost` / `corePort`**：与 `daemon_config.json` 中 `edgeIngress` 一致。核心**晚于**边缘启动时，进程会**自动每约 3 秒重试 TCP**，连上后自动发送 **HELLO**（`currentTreeRevision` 为「核心侧已知 revision」与「本地设备树 revision」的**较大值**），之后 `DATA_RAWBUS` / `TREE_PROPOSE` 才会真正发到核心。 - **`edgeId`**：HELLO 与会话绑定；`TREE_PROPOSE` 必须相同。 -- **`edgeDiscoveryEnabled`**（bool，默认 `true`）：在 TCP 连接且 **HELLO 成功** 后启动 udev 发现协调器，将串口以 `op: upsert` 写入核心设备树（需 Linux + `libudev`）。设为 `false` 可仅依赖手工 `TREE_PROPOSE` 或核心下发树。 -- **`endpoint`**：用于计算 `endpointHash`（与核心一致：`qHash(endpoint)` 即 Qt 默认字符串哈希）。 -- **`fakeDeviceId`**：组装 `routingKey = (endpointHash << 32) | deviceId` 与核心 `makeRoutingKey` 一致。 -- **`fakeIntervalMs`**：设为 `0` 可关闭假数据定时器；**真实串口**由核心下发的设备树节点驱动（`type`/`endpoint` 可识别为 Serial，且 `params.online` 为 true）。节点 `params.deviceId` 会写入上行 `RawBusMessage::deviceId`（未设则 `-1`）。 +- **`edgeHeartbeatIntervalMs`**（int，默认 `5000`，`0` 关闭）：已连接核心 TCP 时，周期性经 **`DEBUG_TUNNEL`** 上送 **`kind: edge_heartbeat`** JSON（含 **`treeRevision`**、**`nodeCount`**、**`treeFingerprint`**（设备树 `tree` 紧凑 JSON 的 `qHash`）、**`coreTreeRevisionKnown`**、时间戳），核心在 `EdgeTcpIngressService` 中打日志便于对账。未设置时可回退读取旧字段 **`fakeIntervalMs`**。 +- **`edgeDiscoveryEnabled`**（bool，默认 `true`）：启动 udev 串口发现。已连接核心且收到有效 **`treeRevision`** 后，将串口以 `op: upsert` 经 **`TREE_PROPOSE`** 同步到核心；**核心不可达或 HELLO 失败**时仍运行发现，并把结果写入**本地** `localTreePath`（与核心上线后的双向同步兼容）。设为 `false` 可仅依赖手工 `TREE_PROPOSE` 或核心下发树。需 Linux + `libudev`。 +- **真实串口**：由设备树节点驱动（`type`/`endpoint` 可识别为 Serial，且 `params.online` 为 true）。节点 `params.deviceId` 会写入上行 `RawBusMessage::deviceId`（未设则 `-1`）；**`endpoint` / `protocol`** 等字段写在设备树节点 `params` 中（与核心约定一致），不再在 manifest 顶层使用假数据字段。 + +## 协议插件与流式解析（与 daemon Pipeline 对齐） + +当 manifest 中 **`plugins.protocols`** 非空时，进程会加载与 `softbus_daemon` **相同 ABI** 的 `.so`（如 `libmodbus_rtu.so`），并在串口 `read` 路径上启用： + +1. **Framer**：优先使用插件内的 `IProtocolFramerSession`（如 Modbus RTU 粘包）；否则退回 **PassthroughFramer**（每块 read 视为一帧，与未启用插件时行为一致）。 +2. **分帧线程与有界队列**：与 daemon `EndpointFramingState` 对齐——按 **`endpointHash`** 各起一条消费线程，串口读回调仅 **`tryEnqueue`** 到 **SPSC `LockFreeQueue`**（容量见 manifest **`edgeSerialChunkQueueCapacity`**，默认 `4096`；**`edgeUseEndpointFramerThreads`** 默认 `true`，`false` 时在同线程同步分帧供调试）。队列满时上报 **`edge stream chunk queue full`**。 +3. **Parser + Mapper**：与核心 `PipelineEngine::stage2Parser` 一致；mapper 在边缘使用 **sniffingEnabled=true** 且无 `mappingLookup`，故无映射规则时仍会产出带 `unmapped` 标记的 DOM 样本（仅 **日志** `EdgeStreamPipeline: parsed DOM ...`，**不上送 DOM 到核心**）。 +4. **上行**：仍只对 **完整帧** 发送 **`DATA_RAWBUS`**（与核心解码路径兼容）。 + +**`plugins.protocols` 路径**：可为绝对路径，或与 manifest 文件同目录的相对路径。请确保边缘运行时能 `dlopen` 到该文件（通常与核心构建产物 `bin/plugins/*.so` 一致）。 ## 运维 HTTP -若 `opsListenPort > 0`，监听 `127.0.0.1:opsListenPort`： +若 `opsListenPort > 0`，在 TCP 连接核心**之前**即启动监听（便于核心未启动时仍能打开本页排查配置）。默认 **`opsListenAllInterfaces` 为 false**：仅绑定 **`127.0.0.1`**，必须在**运行 edge_agent 的本机**浏览器打开。若需从局域网其他电脑访问，在 manifest 中设 **`"opsListenAllInterfaces": true`**（绑定 `0.0.0.0`，请自行评估内网安全）。 - `GET /`：返回 `localTreePath` 文件内容（无文件则返回占位 HTML，含指向配置页的链接）。 - **`GET /ui`**：**阶段 5** 简易配置页——从 **`GET /discovered`** 同源列表选择已发现串口节点，填写 `params.protocol`（如 `modbus_rtu`）、波特率等，提交后由边缘组装 **`op: upsert`** 的 `TREE_PROPOSE` 发往核心（使用当前会话的 **`baseTreeRevision`**，须已 HELLO 且收到过 `TREE_RESULT`）。 @@ -74,12 +87,19 @@ curl -s -X POST http://127.0.0.1:9080/propose \ （`entries` 中节点 id 必须满足核心 `DeviceTreeModel::nodeBelongsToEdge` 规则。） +### 打不开 `/ui` 时排查 + +1. **`opsListenPort` 是否为 0**：为 0 则完全不监听；示例见 `config/edge_manifest.example.json`（如 `9080`）。 +2. **是否用错地址**：默认只能 **`http://127.0.0.1:<端口>/ui`** 在**边缘本机**打开；在别的电脑浏览器里填边缘的 `127.0.0.1` 是无效的，需 **`opsListenAllInterfaces: true`** 后用 **`http://<边缘局域网IP>:<端口>/ui`**，或 SSH 端口转发。 +3. **看进程日志**：启动成功会打印 `ops UI http://127.0.0.1:...` 或 `http://0.0.0.0:...`；若出现 `OpsUiHttp: listen failed`，多为端口被占用。 +4. **核心连不上**：自本版本起，只要 **`opsListenPort > 0`**，即使核心 TCP/HELLO 失败进程也会继续运行并保留运维 HTTP（日志中会提示 `continuing with ops UI only`）；此前版本会在启动 Ops 前退出，表现为「完全打不开」。 + ## 本地树文件 -`TREE_FULL` 落盘格式为 `{ "treeRevision", "edgeId", "tree": [ ... ] }`。**`EdgeDeviceBus`** 会将其中的 `tree` 当作 `device_tree` 加载，并对 **Serial** 与 **CAN**（`endpoint` 形如 `can:vcan0`，`type`/`endpoint` 推断为 CAN）执行 **reconcile**：打开设备后读到的数据经 `EdgeTcpUplinkPort` 编码为 `DATA_RAWBUS` 发往核心。与核心真源同步仍通过 **TREE_PROPOSE**（含边缘 **`upsert`** 自动注册）、核心侧 D-Bus `patch_device_tree_node` 或运维 **`push_edge_tree`**。 +`TREE_FULL` 落盘格式为 `{ "treeRevision", "edgeId", "tree": [ ... ] }`。**`EdgeDeviceBus`** 会将其中的 `tree` 当作 `device_tree` 加载，并对 **Serial** 与 **CAN**（`endpoint` 形如 `can:vcan0`，`type`/`endpoint` 推断为 CAN）执行 **reconcile**：打开设备后读到的数据经 `EdgeTcpUplinkPort` 编码为 `DATA_RAWBUS` 发往核心。串口节点可在 **`params`** 中设置 **`serialReadCoalesceMs`**（0～500，默认 **0**）：读字节先入驱动内缓冲，静音该毫秒数后合并再交给上层；启用流式 pipeline 时主路径为 **每 endpoint 分帧线程 + 有界队列**，一般保持 **0** 即可，仅在仍需驱动侧攒包时再调大。可选 **`serialReadBufferCapBytes`** 限制缓冲上限。 ### 发现 → 注册流程（简要） 1. 边缘连接核心并 **HELLO**；核心下发 **`TREE_FULL`**，本地落盘并 reconcile 已有节点。 -2. `edgeDiscoveryEnabled` 为真时，`EdgeDiscoveryCoordinator` 扫描/监听 udev 串口 **add/remove**，在拿到有效 **`baseTreeRevision`**（来自 `TREE_RESULT`）后发送 **`TREE_PROPOSE`**：`op: upsert` 注册 `edges//discovered/`，remove 时 **`patch`** `params.online: false`。 +2. `edgeDiscoveryEnabled` 为真时，`EdgeDiscoveryCoordinator` 扫描/监听 udev 串口 **add/remove**：若已拿到核心 **`baseTreeRevision`** 则发送 **`TREE_PROPOSE`**；若**无核心**（或 HELLO 未成功）则把 **`op: upsert` / `patch`** 应用到**本地** `DeviceTreeModel` 并落盘 `local_tree.json`，再 **reconcile** 串口。核心恢复后仍通过 **`TREE_PROPOSE`** 与 **`TREE_FULL`** 与真源对齐。 3. 核心 `applyEdgePropose` 成功后再次 **`TREE_FULL`**；边缘 reload 树文件后对 **Serial/CAN** 执行 reconcile，**raw 仅上行**。 diff --git a/config/edge_manifest.example.json b/config/edge_manifest.example.json index 3f88e58..a540cce 100644 --- a/config/edge_manifest.example.json +++ b/config/edge_manifest.example.json @@ -5,10 +5,13 @@ "edgeId": "edge-dev-1", "authToken": "", "opsListenPort": 9080, - "localTreePath": "/tmp/edge_local_tree.json", - "fakeIntervalMs": 5000, - "fakeDeviceId": 1, - "endpoint": "edge:fake", - "protocol": 0, - "edgeDiscoveryEnabled": true + "opsListenAllInterfaces": false, + "localTreePath": "$HOME/softbus/edge/local_tree.json", + "edgeHeartbeatIntervalMs": 5000, + "edgeDiscoveryEnabled": true, + "plugins": { + "protocols": [ + { "path": "../../softbus_daemon/bin/plugins/libmodbus_rtu.so" } + ] + } } diff --git a/config/edge_manifest.schema.json b/config/edge_manifest.schema.json index 97cd155..b43d422 100644 --- a/config/edge_manifest.schema.json +++ b/config/edge_manifest.schema.json @@ -8,15 +8,53 @@ "corePort": { "type": "integer", "description": "daemon_config edgeIngress.port" }, "edgeId": { "type": "string", "description": "HELLO / PROPOSE 与会话绑定的边缘 ID" }, "authToken": { "type": "string", "description": "若核心配置了 edgeIngress.authToken，则此处必须一致" }, - "opsListenPort": { "type": "integer", "minimum": 0, "description": "0 关闭本地运维 HTTP；否则监听 127.0.0.1:port" }, - "localTreePath": { "type": "string", "description": "GET / 返回的本地树文件路径（展示用）" }, - "fakeIntervalMs": { "type": "integer", "minimum": 100, "description": "假数据 DATA_RAWBUS 发送周期（毫秒）" }, - "fakeDeviceId": { "type": "integer", "description": "RawBusMessage.deviceId" }, - "endpoint": { "type": "string", "description": "用于 qHash(endpoint) 得到 endpointHash，须与核心设备树约定一致" }, - "protocol": { "type": "integer", "description": "ProtocolType 枚举数值（与核心 RawBusMessage 一致）" }, + "opsListenPort": { "type": "integer", "minimum": 0, "description": "0 关闭运维 HTTP；否则监听端口。默认仅 127.0.0.1；设 opsListenAllInterfaces 为 true 时监听 0.0.0.0（局域网可访问，注意安全）" }, + "opsListenAllInterfaces": { + "type": "boolean", + "description": "false（默认）时运维 HTTP 仅绑定本机回环；true 时绑定所有网卡，便于从局域网浏览器访问" + }, + "localTreePath": { + "type": "string", + "description": "本地设备树 JSON 路径（Ops GET / 展示、TREE_FULL 落盘）。未配置时内置默认为 $HOME/softbus/edge/local_tree.json。字符串若以 $HOME/、${HOME}/ 或 ~/ 开头，会在加载时展开为当前用户家目录" + }, + "edgeHeartbeatIntervalMs": { + "type": "integer", + "minimum": 0, + "description": "经 DEBUG_TUNNEL 上送边缘心跳的周期（毫秒）。载荷 JSON 含 treeRevision、nodeCount、treeFingerprint（设备树紧凑 JSON 的 qHash）。0 关闭" + }, "edgeDiscoveryEnabled": { "type": "boolean", - "description": "HELLO 成功后是否启动 udev 串口发现并向核心发送 TREE_PROPOSE（upsert/patch）；需 Linux libudev" + "description": "为 true 时启动 udev 串口发现：已连接核心且拿到 treeRevision 后发送 TREE_PROPOSE；无核心或 HELLO 失败时仍发现，并将 upsert/patch 写入本地 local_tree.json 与 reconcile。需 Linux libudev" + }, + "edgeSerialChunkQueueCapacity": { + "type": "integer", + "minimum": 64, + "description": "启用流式 pipeline 时，每个 endpointHash 一条分帧线程对应的有界 SPSC 队列容量（与 daemon EndpointFramingState::chunkQ 语义对齐）。默认 4096" + }, + "edgeUseEndpointFramerThreads": { + "type": "boolean", + "description": "为 true（默认）时串口读回调仅 tryEnqueue 到分帧线程；为 false 时在同线程同步分帧（调试用）。主路径推荐 true" + }, + "plugins": { + "type": "object", + "description": "可选；与 softbus_daemon 的 daemon_config.plugins 形态一致。非空时启用边缘流式 framer/parser/mapper（见 README）", + "properties": { + "protocols": { + "type": "array", + "items": { + "oneOf": [ + { "type": "string", "description": "协议插件 .so 路径（与 manifest 同目录的相对路径可解析）" }, + { + "type": "object", + "required": ["path"], + "properties": { + "path": { "type": "string", "description": "dlopen 路径，通常指向 libmodbus_rtu.so 等" } + } + } + ] + } + } + } } } } diff --git a/src/device_bus/EdgeDeviceBus.cpp b/src/device_bus/EdgeDeviceBus.cpp index 6aba344..2cd6b21 100644 --- a/src/device_bus/EdgeDeviceBus.cpp +++ b/src/device_bus/EdgeDeviceBus.cpp @@ -8,6 +8,7 @@ #include #include "device_bus/manager/SerialDeviceManager.h" +#include "pipeline/EdgeStreamPipeline.h" #if SOFTBUS_HAVE_CAN #include "device_bus/manager/CanDeviceManager.h" #endif @@ -19,7 +20,8 @@ namespace softbus::device_bus EdgeDeviceBus::EdgeDeviceBus() : m_tree(std::make_shared()) - , m_pool(std::make_shared()) + // 多串口并发读 + 分帧再分配：默认 1024 块在 11 路串口下易耗尽，导致 Modbus 分帧静默不出帧 + , m_pool(std::make_shared(4096, 16384)) { } @@ -36,6 +38,20 @@ void EdgeDeviceBus::setIngressPort(softbus::message_bus::ingress::IIngressPort* m_ingress = port; } +void EdgeDeviceBus::setStreamPipeline(std::shared_ptr pipeline) +{ + m_streamPipeline = std::move(pipeline); + if (m_streamPipeline) { + m_streamPipeline->setTreeModel(m_tree.get()); + m_streamPipeline->setFramedSink([this](softbus::core::models::RawBusMessage msg) { + if (!m_ingress) { + return false; + } + return m_ingress->push(std::move(msg)); + }); + } +} + bool EdgeDeviceBus::loadLocalEdgeTreeFile(const QString& path, QString* errorMessage) { QFile f(path); @@ -56,13 +72,18 @@ bool EdgeDeviceBus::loadLocalEdgeTreeFile(const QString& path, QString* errorMes const QJsonObject rootObj = doc.object(); const QJsonArray treeArr = rootObj.value(QStringLiteral("tree")).toArray(); if (treeArr.isEmpty()) { - if (errorMessage) { - *errorMessage = QStringLiteral("missing or empty 'tree' array"); + QJsonObject wrapped; + wrapped.insert(QStringLiteral("device_tree"), QJsonArray()); + if (rootObj.contains(QStringLiteral("treeRevision"))) { + wrapped.insert(QStringLiteral("device_tree_revision"), rootObj.value(QStringLiteral("treeRevision"))); } - return false; + return m_tree->loadFromDaemonConfig(wrapped, errorMessage); } QJsonObject wrapped; wrapped.insert(QStringLiteral("device_tree"), treeArr); + if (rootObj.contains(QStringLiteral("treeRevision"))) { + wrapped.insert(QStringLiteral("device_tree_revision"), rootObj.value(QStringLiteral("treeRevision"))); + } return m_tree->loadFromDaemonConfig(wrapped, errorMessage); } @@ -130,11 +151,16 @@ void EdgeDeviceBus::reconcileSerialFromTree() } } + if (m_streamPipeline) { + m_streamPipeline->setTreeModel(m_tree.get()); + } + softbus::device_bus::manager::SerialInitDeps deps; deps.treeModel = m_tree.get(); deps.dirtyCb = []() {}; deps.ingressPort = m_ingress; deps.memoryPool = m_pool; + deps.streamPipeline = m_streamPipeline; deps.serialDevicesByEndpoint = &m_serialByEndpoint; for (const QString& ep : desired) { diff --git a/src/device_bus/EdgeDeviceBus.h b/src/device_bus/EdgeDeviceBus.h index 2f00e91..2d455c5 100644 --- a/src/device_bus/EdgeDeviceBus.h +++ b/src/device_bus/EdgeDeviceBus.h @@ -14,6 +14,11 @@ #endif #include "message_bus/ingress/IIngressPort.h" +namespace softbus_edge +{ +class EdgeStreamPipeline; +} + namespace softbus::device_bus { @@ -29,6 +34,9 @@ public: void setIngressPort(softbus::message_bus::ingress::IIngressPort* port); + /** 可选：加载与 daemon 相同的协议 .so 后启用 framer+parser+mapper（见 manifest plugins.protocols）。 */ + void setStreamPipeline(std::shared_ptr pipeline); + /** 读取边缘落盘：`{ "tree": [...], "treeRevision", "edgeId" }`，包装为 `loadFromDaemonConfig`。 */ bool loadLocalEdgeTreeFile(const QString& path, QString* errorMessage = nullptr); @@ -51,6 +59,7 @@ private: std::shared_ptr m_tree; std::shared_ptr m_pool; + std::shared_ptr m_streamPipeline; softbus::message_bus::ingress::IIngressPort* m_ingress{nullptr}; QHash> m_serialByEndpoint; #if SOFTBUS_HAVE_CAN diff --git a/src/device_bus/manager/SerialDeviceManager.cpp b/src/device_bus/manager/SerialDeviceManager.cpp index b59a3b1..651755e 100644 --- a/src/device_bus/manager/SerialDeviceManager.cpp +++ b/src/device_bus/manager/SerialDeviceManager.cpp @@ -1,12 +1,16 @@ #include "device_bus/manager/SerialDeviceManager.h" +#include + #include #include #include "device_bus/tree/DeviceTreeModel.h" +#include "core/models/MessageRoutingUtils.h" #include "devices/DeviceTypes.h" #include "hardware/drivers/serial/SerialCapabilities.h" #include "hardware/drivers/serial/SerialQtDriver.h" +#include "pipeline/EdgeStreamPipeline.h" #include "utils/logs/logging.h" using softbus::hardware::drivers::serial::SerialQtDriver; @@ -15,6 +19,27 @@ namespace serial_cap = softbus::hardware::drivers::serial::capabilities; namespace softbus::device_bus::manager { +/** 发现节点常无 params.deviceId；生成稳定正整数供 DATA_RAWBUS / 路由使用。 */ +static int edgeResolveRuntimeDeviceId(const QJsonObject& params, const QString& endpoint) +{ + const int explicitId = params.value(QStringLiteral("deviceId")).toInt(-1); + if (explicitId >= 0) { + return explicitId; + } + const QString sk = params.value(QStringLiteral("stableKey")).toString(); + if (!sk.isEmpty()) { + bool ok = false; + const quint64 u = sk.toULongLong(&ok); + if (ok && u > 0) { + return static_cast((u % 2147483646ULL) + 1ULL); + } + const quint32 h = qHash(sk); + return static_cast(h % 2147483646U) + 1; + } + const quint32 eh = softbus::core::models::makeEndpointHash(endpoint); + return static_cast(eh % 2147483646U) + 1; +} + static bool setIfChanged(QJsonObject& params, const QString& key, const QJsonValue& value) { const auto it = params.find(key); @@ -121,8 +146,15 @@ bool initializeSerialDeviceWithDeps(const QString& endpoint, const QJsonObject& } driver->SetFlowControl(flow); + driver->SetReadCoalesceMs( + std::min(500, std::max(0, params.value(QStringLiteral("serialReadCoalesceMs")).toInt(8)))); + const int capParam = params.value(QStringLiteral("serialReadBufferCapBytes")).toInt(0); + if (capParam > 0) { + driver->SetReadBufferCapBytes(capParam); + } + softbus::devices::DeviceInfo info; - info.id = params.value(QStringLiteral("deviceId")).toInt(-1); + info.id = edgeResolveRuntimeDeviceId(params, endpoint); info.endpoint = endpoint; info.type = QStringLiteral("serial"); info.kind = softbus::devices::DeviceKind::Serial; @@ -139,6 +171,9 @@ bool initializeSerialDeviceWithDeps(const QString& endpoint, const QJsonObject& QSharedPointer device( new softbus::devices::physical::SerialDevice(std::move(driver), cfg, info)); device->bindIngress(deps.ingressPort, deps.memoryPool); + if (deps.streamPipeline) { + device->setStreamPipeline(deps.streamPipeline); + } if (!device->start()) { LOG_ERROR() << "Edge SerialDeviceManager: failed to start serial device for" << endpoint; diff --git a/src/device_bus/manager/SerialDeviceManager.h b/src/device_bus/manager/SerialDeviceManager.h index 30094f6..edaeab9 100644 --- a/src/device_bus/manager/SerialDeviceManager.h +++ b/src/device_bus/manager/SerialDeviceManager.h @@ -13,6 +13,11 @@ #include "devices/physical/SerialDevice.h" #include "message_bus/ingress/IIngressPort.h" +namespace softbus_edge +{ +class EdgeStreamPipeline; +} + class DeviceTreeModel; namespace softbus::device_bus::manager @@ -24,6 +29,8 @@ struct SerialInitDeps std::function dirtyCb; softbus::message_bus::ingress::IIngressPort* ingressPort{nullptr}; std::shared_ptr memoryPool; + /** 可选：与 daemon 对齐的 framer/parser；非空时 SerialDevice 会走流式解析再上行。 */ + std::shared_ptr streamPipeline; QHash>* serialDevicesByEndpoint{nullptr}; }; diff --git a/src/device_bus/monitor/EdgeDiscoveryCoordinator.cpp b/src/device_bus/monitor/EdgeDiscoveryCoordinator.cpp index 5c75bed..81aa394 100644 --- a/src/device_bus/monitor/EdgeDiscoveryCoordinator.cpp +++ b/src/device_bus/monitor/EdgeDiscoveryCoordinator.cpp @@ -5,8 +5,11 @@ #include #include #include +#include #include +#include "device_bus/EdgeDeviceBus.h" +#include "device_bus/tree/DeviceTreeModel.h" #include "device_bus/monitor/UdevDeviceMonitor.h" #include "message_bus/ingress/EdgeSession.h" #include "utils/logs/logging.h" @@ -39,6 +42,18 @@ void EdgeDiscoveryCoordinator::setCoreTreeRevisionGetter(std::function m_getCoreRevision = std::move(fn); } +void EdgeDiscoveryCoordinator::setLocalEdgePersistence(softbus::device_bus::EdgeDeviceBus* bus, + const QString& localTreePath) +{ + m_edgeBus = bus; + m_localTreePath = localTreePath; +} + +void EdgeDiscoveryCoordinator::setStandaloneLocalOnly(bool standalone) +{ + m_standaloneLocalOnly = standalone; +} + QString EdgeDiscoveryCoordinator::stableKeyForSerial(const softbus::device_bus::monitor::DeviceEvent& ev) const { const QString syspath = ev.props.value(QStringLiteral("SYSPATH")); @@ -63,7 +78,7 @@ void EdgeDiscoveryCoordinator::enqueueUpsert(const softbus::device_bus::monitor: params.insert(QStringLiteral("edgeId"), m_edgeId); params.insert(QStringLiteral("stableKey"), sk); params.insert(QStringLiteral("online"), true); - params.insert(QStringLiteral("baudRate"), 115200); + params.insert(QStringLiteral("baudRate"), 9600); params.insert(QStringLiteral("protocol"), QStringLiteral("modbus_rtu")); params.insert(QStringLiteral("discoveredVia"), QStringLiteral("udev")); if (ev.props.contains(QStringLiteral("SYSPATH"))) { @@ -161,17 +176,12 @@ void EdgeDiscoveryCoordinator::scheduleFlush() void EdgeDiscoveryCoordinator::flushPending() { - if (!m_session || m_edgeId.isEmpty()) { + if (m_edgeId.isEmpty()) { return; } if (m_proposeInFlight) { return; } - const qint64 base = m_getCoreRevision ? m_getCoreRevision() : 0; - if (base <= 0) { - LOG_INFO() << "EdgeDiscoveryCoordinator: skip flush, core treeRevision not ready yet"; - return; - } if (m_pendingUpsert.isEmpty() && m_pendingPatchEntries.isEmpty()) { return; } @@ -192,18 +202,88 @@ void EdgeDiscoveryCoordinator::flushPending() return; } - QJsonObject body; - body.insert(QStringLiteral("edgeId"), m_edgeId); - body.insert(QStringLiteral("baseTreeRevision"), base); - body.insert(QStringLiteral("entries"), entries); + const qint64 coreRev = m_getCoreRevision ? m_getCoreRevision() : 0; + const bool useCorePropose = !m_standaloneLocalOnly && coreRev > 0 && m_session; - if (!m_session->sendTreePropose(body)) { - LOG_WARNING() << "EdgeDiscoveryCoordinator: sendTreePropose failed"; + if (useCorePropose) { + QJsonObject body; + body.insert(QStringLiteral("edgeId"), m_edgeId); + body.insert(QStringLiteral("baseTreeRevision"), coreRev); + body.insert(QStringLiteral("entries"), entries); + + if (!m_session->sendTreePropose(body)) { + LOG_WARNING() << "EdgeDiscoveryCoordinator: sendTreePropose failed"; + return; + } + + m_proposeInFlight = true; + LOG_INFO() << "EdgeDiscoveryCoordinator: sent TREE_PROPOSE" << entries.size() << "entries baseRev" << coreRev; return; } - m_proposeInFlight = true; - LOG_INFO() << "EdgeDiscoveryCoordinator: sent TREE_PROPOSE" << entries.size() << "entries baseRev" << base; + if (!m_standaloneLocalOnly && coreRev <= 0) { + LOG_INFO() << "EdgeDiscoveryCoordinator: skip flush, waiting for core treeRevision (HELLO / TREE_RESULT)"; + return; + } + + if (!m_edgeBus) { + LOG_WARNING() << "EdgeDiscoveryCoordinator: standalone/local mode but EdgeDeviceBus not set, cannot persist"; + return; + } + + DeviceTreeModel& tm = m_edgeBus->treeModel(); + const qint64 localBase = tm.treeRevision(); + const DeviceTreeModel::EdgeProposeResult res = tm.applyEdgePropose(m_edgeId, localBase, entries); + if (!res.ok) { + LOG_WARNING() << "EdgeDiscoveryCoordinator: local applyEdgePropose failed" << res.code << res.message; + if (res.code == QStringLiteral("CONFLICT")) { + scheduleFlush(); + } + return; + } + + QString perr; + if (!persistLocalEdgeTree(&perr)) { + LOG_WARNING() << "EdgeDiscoveryCoordinator: persist local tree failed" << perr; + } + m_edgeBus->reconcileSerialFromTree(); + m_edgeBus->reconcileCanFromTree(); + + m_pendingUpsert.clear(); + m_pendingPatchEntries.clear(); + LOG_INFO() << "EdgeDiscoveryCoordinator: applied discovery to local tree revision" << res.newRevision << "nodes" + << tm.nodes().size(); +} + +bool EdgeDiscoveryCoordinator::persistLocalEdgeTree(QString* errOut) +{ + if (!m_edgeBus || m_localTreePath.isEmpty()) { + if (errOut) { + *errOut = QStringLiteral("not configured"); + } + return false; + } + const DeviceTreeModel& tm = m_edgeBus->treeModel(); + QJsonObject doc; + doc.insert(QStringLiteral("treeRevision"), tm.treeRevision()); + doc.insert(QStringLiteral("edgeId"), m_edgeId); + doc.insert(QStringLiteral("tree"), tm.toDeviceTreeJson()); + + QSaveFile out(m_localTreePath); + if (!out.open(QIODevice::WriteOnly | QIODevice::Truncate)) { + if (errOut) { + *errOut = QStringLiteral("open failed"); + } + return false; + } + out.write(QJsonDocument(doc).toJson(QJsonDocument::Indented)); + if (!out.commit()) { + if (errOut) { + *errOut = QStringLiteral("commit failed"); + } + return false; + } + return true; } void EdgeDiscoveryCoordinator::onTreeResultFromCore(const QJsonObject& o) @@ -283,4 +363,9 @@ void EdgeDiscoveryCoordinator::stop() m_devnodeToStableKey.clear(); } +void EdgeDiscoveryCoordinator::requestFlush() +{ + scheduleFlush(); +} + } // namespace softbus_edge diff --git a/src/device_bus/monitor/EdgeDiscoveryCoordinator.h b/src/device_bus/monitor/EdgeDiscoveryCoordinator.h index 79afac5..4be64f2 100644 --- a/src/device_bus/monitor/EdgeDiscoveryCoordinator.h +++ b/src/device_bus/monitor/EdgeDiscoveryCoordinator.h @@ -25,10 +25,15 @@ namespace softbus::message_bus::ingress class EdgeSession; } +namespace softbus::device_bus +{ +class EdgeDeviceBus; +} + namespace softbus_edge { -/** udev 串口发现 → 组装 `TREE_PROPOSE`（含 `op: upsert`）向核心注册设备树节点。 */ +/** udev 串口发现 → 组装 `TREE_PROPOSE` 向核心同步；无核心时写入本地 `DeviceTreeModel` 并落盘 `localTreePath`。 */ class EdgeDiscoveryCoordinator final : public QObject { public: @@ -39,9 +44,18 @@ public: void setEdgeId(const QString& edgeId); void setCoreTreeRevisionGetter(std::function fn); + /** 无核心或尚未拿到 core treeRevision 时，将发现结果写入本地 `DeviceTreeModel` 并落盘到 `localTreePath`。 */ + void setLocalEdgePersistence(softbus::device_bus::EdgeDeviceBus* bus, const QString& localTreePath); + + /** 由 main 设置：TCP/HELLO 未就绪、仅以边缘为真源运行（udev → 本地树）时为 true。 */ + void setStandaloneLocalOnly(bool standalone); + bool start(); void stop(); + /** 请求在防抖后尽快执行一次 `flushPending`（例如核心链路刚恢复时）。 */ + void requestFlush(); + void onTreeResultFromCore(const QJsonObject& o); QJsonArray recentDiscoveriesJson() const; @@ -54,10 +68,14 @@ private: void enqueueOfflinePatch(const QString& devnode); void scheduleFlush(); void flushPending(); + bool persistLocalEdgeTree(QString* errOut); softbus::message_bus::ingress::EdgeSession* m_session{nullptr}; QString m_edgeId; std::function m_getCoreRevision; + softbus::device_bus::EdgeDeviceBus* m_edgeBus{nullptr}; + QString m_localTreePath; + bool m_standaloneLocalOnly{false}; std::unique_ptr m_udev; diff --git a/src/devices/physical/SerialDevice.cpp b/src/devices/physical/SerialDevice.cpp index f9aec83..5606d22 100644 --- a/src/devices/physical/SerialDevice.cpp +++ b/src/devices/physical/SerialDevice.cpp @@ -11,6 +11,7 @@ #include "core/models/RawBusMessage.h" #include "hardware/drivers/serial/SerialQtDriver.h" #include "message_bus/ingress/IIngressPort.h" +#include "pipeline/EdgeStreamPipeline.h" #include "utils/logs/logging.h" using softbus::hardware::drivers::serial::SerialQtDriver; @@ -78,6 +79,11 @@ void SerialDevice::bindIngress(softbus::message_bus::ingress::IIngressPort* ingr m_pool = std::move(pool); } +void SerialDevice::setStreamPipeline(std::shared_ptr pipeline) +{ + m_streamPipeline = std::move(pipeline); +} + // 串口写入数据回调 不定长数据 bool SerialDevice::writePayload(const std::vector& data) { @@ -111,24 +117,28 @@ void SerialDevice::onRead(const std::vector& data) std::copy(data.begin(), data.end(), block->data); block->size = data.size(); - // @test code LOG_INFO() << "SerialDevice: onRead data size" << data.size() << "endpoint" << m_config.endpoint; - // 将数据推送到 ingress 队列 包含端口名、协议、数据、时间戳、traceId - softbus::core::models::RawBusMessage ctx; - ctx.direction = softbus::core::models::BusDirection::Upstream; - ctx.endpointHash = softbus::core::models::makeEndpointHash(m_config.endpoint); - ctx.protocol = softbus::core::models::protocolTypeFromHint(m_info.protocol); - ctx.logicalAddress = 0; - ctx.routingKey = softbus::core::models::makeRoutingKey(ctx.endpointHash, m_info.id); - ctx.timestampMs = QDateTime::currentMSecsSinceEpoch(); - ctx.deviceId = m_info.id; - ctx.payload.block = std::move(block); - ctx.payload.length = data.size(); - ctx.traceId = QStringLiteral("serial:%1:%2") - .arg(m_config.endpoint) - .arg(ctx.timestampMs); - if (!m_ingress->push(std::move(ctx))) { + softbus::core::models::RawBusMessage chunk; + chunk.direction = softbus::core::models::BusDirection::Upstream; + chunk.endpointHash = softbus::core::models::makeEndpointHash(m_config.endpoint); + chunk.protocol = softbus::core::models::protocolTypeFromHint(m_info.protocol); + chunk.logicalAddress = 0; + chunk.routingKey = softbus::core::models::makeRoutingKey(chunk.endpointHash, m_info.id); + chunk.timestampMs = QDateTime::currentMSecsSinceEpoch(); + chunk.deviceId = m_info.id; + chunk.payload.block = std::move(block); + chunk.payload.length = data.size(); + chunk.traceId = QStringLiteral("serial:%1:%2").arg(m_config.endpoint).arg(chunk.timestampMs); + + if (m_streamPipeline && m_streamPipeline->ok()) { + if (!m_streamPipeline->tryEnqueueStreamChunk(std::move(chunk))) { + m_ingress->reportError(m_config.endpoint, QStringLiteral("edge stream chunk queue full")); + } + return; + } + + if (!m_ingress->push(std::move(chunk))) { m_ingress->reportError(m_config.endpoint, QStringLiteral("ingress queue full")); } } diff --git a/src/devices/physical/SerialDevice.h b/src/devices/physical/SerialDevice.h index 52a1795..90011ec 100644 --- a/src/devices/physical/SerialDevice.h +++ b/src/devices/physical/SerialDevice.h @@ -10,6 +10,7 @@ namespace softbus::hardware::drivers::serial { class SerialQtDriver; } namespace softbus::message_bus::ingress { class IIngressPort; } namespace softbus::core::memory { class MemoryPool; } +namespace softbus_edge { class EdgeStreamPipeline; } namespace softbus::devices::physical { @@ -42,6 +43,8 @@ public: softbus::devices::DeviceKind kind() const; void bindIngress(softbus::message_bus::ingress::IIngressPort* ingress, std::shared_ptr pool); + /** 可选：与 daemon Pipeline 对齐的 framer+parser+mapper；未设置则保持每 read 直接上行。 */ + void setStreamPipeline(std::shared_ptr pipeline); bool writePayload(const std::vector& data); bool writePayload(const std::uint8_t* data, std::size_t size); @@ -55,6 +58,7 @@ private: softbus::devices::DeviceInfo m_info; softbus::message_bus::ingress::IIngressPort* m_ingress{nullptr}; std::shared_ptr m_pool; + std::shared_ptr m_streamPipeline; }; } // namespace softbus::devices::physical diff --git a/src/fake_driver.cpp b/src/fake_driver.cpp deleted file mode 100644 index b4d850b..0000000 --- a/src/fake_driver.cpp +++ /dev/null @@ -1,20 +0,0 @@ -#include "fake_driver.hpp" - -FakeDriver::FakeDriver(QObject* parent) - : QObject(parent) -{ - connect(&m_timer, &QTimer::timeout, this, &FakeDriver::onTimeout); -} - -void FakeDriver::start(int intervalMs, std::function onTick) -{ - m_onTick = std::move(onTick); - m_timer.start(intervalMs); -} - -void FakeDriver::onTimeout() -{ - if (m_onTick) { - m_onTick(); - } -} diff --git a/src/fake_driver.hpp b/src/fake_driver.hpp deleted file mode 100644 index 06420d5..0000000 --- a/src/fake_driver.hpp +++ /dev/null @@ -1,22 +0,0 @@ -#pragma once - -#include - -#include -#include - -class FakeDriver final : public QObject -{ - Q_OBJECT -public: - explicit FakeDriver(QObject* parent = nullptr); - - void start(int intervalMs, std::function onTick); - -private slots: - void onTimeout(); - -private: - QTimer m_timer; - std::function m_onTick; -}; diff --git a/src/hardware/drivers/serial/SerialQtDriver.cpp b/src/hardware/drivers/serial/SerialQtDriver.cpp index 58c6e0f..aaaed21 100644 --- a/src/hardware/drivers/serial/SerialQtDriver.cpp +++ b/src/hardware/drivers/serial/SerialQtDriver.cpp @@ -1,7 +1,12 @@ #include "hardware/drivers/serial/SerialQtDriver.h" +#include +#include + #include #include +#include +#include #include #include "utils/logs/logging.h" @@ -13,6 +18,10 @@ SerialQtDriver::SerialQtDriver(QObject* parent) : QObject(parent), m_port(this) { + m_flushReadCoalesce = new QTimer(this); + m_flushReadCoalesce->setSingleShot(true); + m_flushReadCoalesce->setTimerType(Qt::PreciseTimer); + connect(m_flushReadCoalesce, &QTimer::timeout, this, &SerialQtDriver::flushPendingRead); connect(&m_port, &QSerialPort::readyRead, this, &SerialQtDriver::handleReadyRead); connect(&m_port, &QSerialPort::errorOccurred, this, &SerialQtDriver::handleError); } @@ -106,23 +115,53 @@ void SerialQtDriver::SetFlowControl(QSerialPort::FlowControl flow) m_port.setFlowControl(flow); } +void SerialQtDriver::SetReadCoalesceMs(int ms) +{ + m_readCoalesceMs = std::max(0, ms); +} + +void SerialQtDriver::SetReadBufferCapBytes(int cap) +{ + m_readBufferCap = std::max(1024, cap); +} + void SerialQtDriver::handleReadyRead() { + const QByteArray data = m_port.readAll(); + if (data.isEmpty()) { + return; + } if (!m_readCallback) { - const QByteArray data = m_port.readAll(); LOG_DEBUG() << "SerialQtDriver: dropped" << data.size() << "bytes on" << m_port.portName() << "because no read callback is set"; return; } - const QByteArray data = m_port.readAll(); - if (data.isEmpty()) { + m_pendingReadBuf.append(data); + if (m_pendingReadBuf.size() >= m_readBufferCap) { + flushPendingRead(); return; } - std::vector buffer(reinterpret_cast(data.constData()), - reinterpret_cast(data.constData()) + data.size()); - m_readCallback(buffer); + if (m_readCoalesceMs <= 0) { + flushPendingRead(); + return; + } + + m_flushReadCoalesce->start(m_readCoalesceMs); +} + +void SerialQtDriver::flushPendingRead() +{ + if (!m_readCallback || m_pendingReadBuf.isEmpty()) { + m_pendingReadBuf.clear(); + return; + } + const int n = m_pendingReadBuf.size(); + std::vector buffer(static_cast(n)); + std::memcpy(buffer.data(), m_pendingReadBuf.constData(), static_cast(n)); + m_pendingReadBuf.clear(); + m_readCallback(std::move(buffer)); } void SerialQtDriver::handleError(QSerialPort::SerialPortError error) @@ -170,12 +209,27 @@ bool SerialQtDriver::openOnOwnerThread(const QString& endpoint) return false; } + // 显式接收缓冲：避免默认过小导致高频小 read；具体上限仍受 OS 驱动约束。 + m_port.setReadBufferSize(256 * 1024); + m_pendingReadBuf.clear(); + if (m_flushReadCoalesce) { + m_flushReadCoalesce->stop(); + } + LOG_DEBUG() << "SerialQtDriver: opened port" << m_port.portName(); return true; } void SerialQtDriver::closeOnOwnerThread() { + if (m_flushReadCoalesce) { + m_flushReadCoalesce->stop(); + } + if (!m_pendingReadBuf.isEmpty()) { + flushPendingRead(); + } else { + m_pendingReadBuf.clear(); + } if (m_port.isOpen()) { const QString name = m_port.portName(); m_port.close(); diff --git a/src/hardware/drivers/serial/SerialQtDriver.h b/src/hardware/drivers/serial/SerialQtDriver.h index 8db4c07..f8f8c69 100644 --- a/src/hardware/drivers/serial/SerialQtDriver.h +++ b/src/hardware/drivers/serial/SerialQtDriver.h @@ -1,10 +1,12 @@ #pragma once +#include #include #include #include #include -#include + +class QTimer; #include "hardware/interfaces/IStreamDriver.h" @@ -74,19 +76,32 @@ public: void SetStopBits(QSerialPort::StopBits stopBits); void SetFlowControl(QSerialPort::FlowControl flow); + /** + * 读侧字节合并（对齐 daemon 侧 Pipeline 入队前的「攒包」思路，减轻高频 1~2 字节 readyRead 对上层/framer 的冲击）。 + * 0：每次 `readAll()` 后立即回调；>0：数据先入内部缓冲，连续 `readyRead` 静默达到 ms 后合并为一次回调。默认 0ms。 + */ + void SetReadCoalesceMs(int ms); + /** 内部读缓冲上限（字节），达到后强制立刻 flush，避免无限增长。默认 256KiB。 */ + void SetReadBufferCapBytes(int cap); + private slots: void handleReadyRead(); void handleError(QSerialPort::SerialPortError error); + void flushPendingRead(); private: bool openOnOwnerThread(const QString& endpoint); void closeOnOwnerThread(); bool writeOnOwnerThread(const QByteArray& bytes); -private: QSerialPort m_port; ReadCallback m_readCallback; ErrorCallback m_errorCallback; + + QTimer* m_flushReadCoalesce{nullptr}; + QByteArray m_pendingReadBuf; + int m_readCoalesceMs{0}; + int m_readBufferCap{256 * 1024}; }; } // namespace softbus::hardware::drivers::serial diff --git a/src/main.cpp b/src/main.cpp index 31ca86b..19bed47 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -1,22 +1,56 @@ #include #include #include +#include #include +#include #include #include #include #include #include +#include + +#include #include "device_bus/EdgeDeviceBus.h" #include "device_bus/monitor/EdgeDiscoveryCoordinator.h" #include "message_bus/ingress/EdgeSession.h" #include "message_bus/ingress/EdgeTcpUplinkPort.h" -#include "message_bus/ingress/EdgeUplinkWire.h" +#include "pipeline/EdgeStreamPipeline.h" -#include "fake_driver.hpp" #include "ops_ui_http.hpp" +/** 默认数据目录：`$HOME/softbus/edge`（自动创建）。 */ +static QString edgeDefaultDataDir() +{ + return QDir::cleanPath(QDir::homePath() + QLatin1String("/softbus/edge")); +} + +static QString edgeDefaultManifestPath() +{ + return edgeDefaultDataDir() + QLatin1String("/edge_manifest.json"); +} + +static QString edgeDefaultLocalTreePath() +{ + return edgeDefaultDataDir() + QLatin1String("/local_tree.json"); +} + +/** manifest 中 `localTreePath` 若以 `$HOME/`、`${HOME}/` 或 `~/` 开头，则展开为当前用户家目录。 */ +static QString expandManifestHomePath(QString p) +{ + const QString home = QDir::homePath(); + if (p.startsWith(QLatin1String("$HOME/"))) { + p.replace(0, 6, home); + } else if (p.startsWith(QLatin1String("${HOME}/"))) { + p.replace(0, 7, home); + } else if (p.startsWith(QLatin1String("~/"))) { + p.replace(0, 1, home); + } + return QDir::cleanPath(p); +} + /** 无配置文件或与核心同目录运行时使用的默认值（可被 manifest 覆盖）。 */ static QJsonObject defaultManifest() { @@ -26,12 +60,13 @@ static QJsonObject defaultManifest() m.insert(QStringLiteral("edgeId"), QStringLiteral("edge-dev-1")); m.insert(QStringLiteral("authToken"), QStringLiteral("")); m.insert(QStringLiteral("opsListenPort"), 0); - m.insert(QStringLiteral("localTreePath"), QStringLiteral("/tmp/edge_local_tree.json")); - m.insert(QStringLiteral("fakeIntervalMs"), 5000); - m.insert(QStringLiteral("fakeDeviceId"), 1); - m.insert(QStringLiteral("endpoint"), QStringLiteral("edge:fake")); - m.insert(QStringLiteral("protocol"), 0); + m.insert(QStringLiteral("opsListenAllInterfaces"), false); + m.insert(QStringLiteral("localTreePath"), edgeDefaultLocalTreePath()); + m.insert(QStringLiteral("edgeHeartbeatIntervalMs"), 5000); m.insert(QStringLiteral("edgeDiscoveryEnabled"), true); + QJsonObject plugins; + plugins.insert(QStringLiteral("protocols"), QJsonArray()); + m.insert(QStringLiteral("plugins"), plugins); return m; } @@ -57,6 +92,170 @@ static QJsonObject loadManifestOrDefaults(const QString& path) return out; } +/** 创建 `$HOME/softbus/edge`，并写入首次启动的 `edge_manifest.json` / `local_tree.json`（若不存在）。 */ +static bool ensureEdgeDataLayout(QString* errOut) +{ + const QString root = edgeDefaultDataDir(); + QDir dir; + if (!dir.mkpath(root)) { + if (errOut) { + *errOut = QStringLiteral("mkpath failed: %1").arg(root); + } + return false; + } + const QString treePath = edgeDefaultLocalTreePath(); + if (!QFile::exists(treePath)) { + QJsonObject doc; + doc.insert(QStringLiteral("treeRevision"), 0); + doc.insert(QStringLiteral("edgeId"), QStringLiteral("edge-dev-1")); + doc.insert(QStringLiteral("tree"), QJsonArray()); + QFile f(treePath); + if (!f.open(QIODevice::WriteOnly | QIODevice::Truncate)) { + if (errOut) { + *errOut = QStringLiteral("cannot write %1").arg(treePath); + } + return false; + } + f.write(QJsonDocument(doc).toJson(QJsonDocument::Indented)); + } + const QString manPath = edgeDefaultManifestPath(); + if (!QFile::exists(manPath)) { + QFile f(manPath); + if (!f.open(QIODevice::WriteOnly | QIODevice::Truncate)) { + if (errOut) { + *errOut = QStringLiteral("cannot write %1").arg(manPath); + } + return false; + } + f.write(QJsonDocument(defaultManifest()).toJson(QJsonDocument::Indented)); + } + return true; +} + +static QStringList protocolPluginPathsFromManifest(const QJsonObject& manifest, const QString& manifestPath) +{ + QStringList out; + const QJsonArray arr = + manifest.value(QStringLiteral("plugins")).toObject().value(QStringLiteral("protocols")).toArray(); + const QFileInfo mf(manifestPath); + const QDir baseDir = mf.absoluteDir(); + for (const QJsonValue& v : arr) { + QString p; + if (v.isObject()) { + p = v.toObject().value(QStringLiteral("path")).toString(); + } else if (v.isString()) { + p = v.toString(); + } + if (p.isEmpty()) { + continue; + } + if (!QFileInfo(p).isAbsolute()) { + p = baseDir.filePath(p); + } + out.append(p); + } + return out; +} + +static bool appendPluginPathIfFileExists(QStringList& paths, const QString& candidate) +{ + const QFileInfo fi(candidate); + if (!fi.exists() || !fi.isFile()) { + return false; + } + paths.append(fi.absoluteFilePath()); + return true; +} + +/** 从 `startDir` 起向父目录最多爬 `maxLevels` 层，查找 `softbus_daemon/bin/plugins/libmodbus_rtu.so`。 */ +static bool appendDefaultModbusPluginByWalkingUp(QStringList& paths, const QString& startDir, int maxLevels = 10) +{ + QDir d(QFileInfo(startDir).absoluteFilePath()); + if (!d.exists()) { + return false; + } + for (int i = 0; i < maxLevels; ++i) { + const QString candidate = + d.absoluteFilePath(QStringLiteral("softbus_daemon/bin/plugins/libmodbus_rtu.so")); + if (appendPluginPathIfFileExists(paths, candidate)) { + qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded" << paths.constLast(); + return true; + } + if (!d.cdUp()) { + break; + } + } + return false; +} + +/** manifest 未配置 plugins 时，尝试从常见相对路径加载 modbus_rtu（与仓库内 edge_manifest.example 一致）。 */ +static void appendDefaultProtocolPluginsIfManifestEmpty(QStringList& paths, const QString& manifestPath) +{ + if (!paths.isEmpty()) { + return; + } + const QByteArray envPath = qgetenv("EDGE_MODBUS_RTU_SO"); + if (!envPath.isEmpty()) { + const QString p = QString::fromLocal8Bit(envPath.trimmed()); + if (appendPluginPathIfFileExists(paths, p)) { + qInfo() << "Edge: loaded protocol plugin from EDGE_MODBUS_RTU_SO" << paths.constLast(); + return; + } + qWarning() << "Edge: EDGE_MODBUS_RTU_SO is set but file not found or not a file:" << p; + } + + // 与数据目录约定一致：$HOME/softbus/bin/plugins 放置协议插件（用户态安装） + const QString homeSoftbusPlugin = + QDir(QDir::homePath()).filePath(QStringLiteral("softbus/bin/plugins/libmodbus_rtu.so")); + if (appendPluginPathIfFileExists(paths, homeSoftbusPlugin)) { + qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded" << paths.constLast(); + return; + } + + const QString appBin = QCoreApplication::applicationDirPath(); + const QString manDir = QFileInfo(manifestPath).absolutePath(); + const QString exeFile = QCoreApplication::applicationFilePath(); + + if (!exeFile.isEmpty() && appendDefaultModbusPluginByWalkingUp(paths, QFileInfo(exeFile).absolutePath())) { + return; + } + if (appendDefaultModbusPluginByWalkingUp(paths, appBin)) { + return; + } + if (appendDefaultModbusPluginByWalkingUp(paths, manDir)) { + return; + } + + const QStringList legacyCandidates = { + QDir(appBin).filePath(QStringLiteral("plugins/libmodbus_rtu.so")), + QDir(appBin).filePath(QStringLiteral("../../../softbus_daemon/bin/plugins/libmodbus_rtu.so")), + QDir(manDir).filePath(QStringLiteral("../../softbus_daemon/bin/plugins/libmodbus_rtu.so")), + QDir(manDir).filePath(QStringLiteral("../softbus_daemon/bin/plugins/libmodbus_rtu.so")), + }; + for (const QString& p : legacyCandidates) { + if (appendPluginPathIfFileExists(paths, p)) { + qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded" << paths.constLast(); + return; + } + } + + const QString home = QDir::homePath(); + const QStringList homeDevCandidates = { + QDir(home).filePath(QStringLiteral("myCode/bus/soft_bus/softbus_daemon/bin/plugins/libmodbus_rtu.so")), + QDir(home).filePath(QStringLiteral("myCode/bus/soft_bus/softbus_daemon/build/plugins/libmodbus_rtu.so")), + }; + for (const QString& p : homeDevCandidates) { + if (appendPluginPathIfFileExists(paths, p)) { + qInfo() << "Edge: manifest has no plugins.protocols; auto-loaded (under $HOME)" << paths.constLast(); + return; + } + } + + qWarning() << "Edge: plugins.protocols is empty and libmodbus_rtu.so not found. Expected" + " $HOME/softbus/bin/plugins/libmodbus_rtu.so (or set EDGE_MODBUS_RTU_SO / add plugins to" + " edge_manifest.json)."; +} + static void tryReloadTreeFromDisk(softbus::device_bus::EdgeDeviceBus& bus, const QString& localTreePath) { QString err; @@ -69,11 +268,47 @@ static void tryReloadTreeFromDisk(softbus::device_bus::EdgeDeviceBus& bus, const } } +/** 兼容旧 manifest 的 `fakeIntervalMs`。 */ +static int edgeHeartbeatIntervalMsFromManifest(const QJsonObject& m) +{ + if (m.contains(QStringLiteral("edgeHeartbeatIntervalMs"))) { + return m.value(QStringLiteral("edgeHeartbeatIntervalMs")).toInt(5000); + } + return m.value(QStringLiteral("fakeIntervalMs")).toInt(5000); +} + +/** 经 `DEBUG_TUNNEL` 上送：设备树节点数、revision、紧凑 JSON 的 qHash 指纹（便于对账）。 */ +static QJsonObject makeEdgeHeartbeatPayload(const QString& edgeId, + qint64 coreTreeRevisionKnown, + const softbus::device_bus::EdgeDeviceBus& bus) +{ + const DeviceTreeModel& tm = bus.treeModel(); + const QJsonArray treeJson = tm.toDeviceTreeJson(); + const QByteArray canon = QJsonDocument(treeJson).toJson(QJsonDocument::Compact); + const quint32 treeFp = static_cast(qHash(canon)); + + QJsonObject o; + o.insert(QStringLiteral("kind"), QStringLiteral("edge_heartbeat")); + o.insert(QStringLiteral("edgeId"), edgeId); + o.insert(QStringLiteral("tsMs"), QDateTime::currentMSecsSinceEpoch()); + o.insert(QStringLiteral("treeRevision"), static_cast(tm.treeRevision())); + o.insert(QStringLiteral("nodeCount"), treeJson.size()); + o.insert(QStringLiteral("treeFingerprint"), static_cast(treeFp)); + o.insert(QStringLiteral("coreTreeRevisionKnown"), coreTreeRevisionKnown); + return o; +} + int main(int argc, char* argv[]) { QCoreApplication app(argc, argv); - const QString manifestPath = argc > 1 ? QString::fromLocal8Bit(argv[1]) - : QStringLiteral("config/edge_manifest.json"); + + QString layoutErr; + if (!ensureEdgeDataLayout(&layoutErr)) { + qWarning() << "ensureEdgeDataLayout:" << layoutErr; + } + + const QString manifestPath = + argc > 1 ? QString::fromLocal8Bit(argv[1]) : edgeDefaultManifestPath(); const QJsonObject m = loadManifestOrDefaults(manifestPath); const QString coreHost = m.value(QStringLiteral("coreHost")).toString(QStringLiteral("127.0.0.1")); @@ -81,11 +316,10 @@ int main(int argc, char* argv[]) const QString edgeId = m.value(QStringLiteral("edgeId")).toString(QStringLiteral("edge-dev-1")); const QString authToken = m.value(QStringLiteral("authToken")).toString(); const int opsPort = m.value(QStringLiteral("opsListenPort")).toInt(0); - const QString localTreePath = m.value(QStringLiteral("localTreePath")).toString(QStringLiteral("/tmp/edge_local_tree.json")); - const int fakeIntervalMs = m.value(QStringLiteral("fakeIntervalMs")).toInt(5000); - const int deviceId = m.value(QStringLiteral("fakeDeviceId")).toInt(1); - const QString endpoint = m.value(QStringLiteral("endpoint")).toString(QStringLiteral("edge:fake")); - const int protocol = m.value(QStringLiteral("protocol")).toInt(0); + const bool opsListenAll = m.value(QStringLiteral("opsListenAllInterfaces")).toBool(false); + const QString localTreePath = + expandManifestHomePath(m.value(QStringLiteral("localTreePath")).toString(edgeDefaultLocalTreePath())); + const int heartbeatIntervalMs = edgeHeartbeatIntervalMsFromManifest(m); const bool edgeDiscoveryEnabled = m.value(QStringLiteral("edgeDiscoveryEnabled")).toBool(true); softbus::device_bus::EdgeDeviceBus edgeBus; @@ -96,11 +330,35 @@ int main(int argc, char* argv[]) discovery.setEdgeSession(&session); discovery.setEdgeId(edgeId); discovery.setCoreTreeRevisionGetter([&coreTreeRevision]() { return coreTreeRevision; }); + discovery.setLocalEdgePersistence(&edgeBus, localTreePath); auto tcpUplink = std::make_unique( [&session](const QByteArray& fr) { return session.writeSbupFrame(fr); }); edgeBus.setIngressPort(tcpUplink.get()); + QStringList pluginPaths = protocolPluginPathsFromManifest(m, manifestPath); + appendDefaultProtocolPluginsIfManifestEmpty(pluginPaths, manifestPath); + { + softbus_edge::EdgeStreamPipeline::Config pcfg; + pcfg.pool = edgeBus.memoryPool(); + pcfg.protocolPluginPaths = pluginPaths; + pcfg.treeModel = &edgeBus.treeModel(); + pcfg.edgeSerialChunkQueueCapacity = + static_cast(std::max(64, m.value(QStringLiteral("edgeSerialChunkQueueCapacity")).toInt(4096))); + pcfg.edgeUseEndpointFramerThreads = m.value(QStringLiteral("edgeUseEndpointFramerThreads")).toBool(true); + auto pipe = std::make_shared(pcfg); + if (!pipe->ok()) { + qWarning() << "EdgeStreamPipeline: disabled:" << pipe->lastError(); + } else { + edgeBus.setStreamPipeline(pipe); + if (!pluginPaths.isEmpty()) { + qInfo() << "EdgeStreamPipeline: loaded protocol plugins" << pluginPaths; + } else { + qInfo() << "EdgeStreamPipeline: active (passthrough; manifest has no protocol plugins)"; + } + } + } + if (QFile::exists(localTreePath)) { tryReloadTreeFromDisk(edgeBus, localTreePath); } @@ -146,19 +404,10 @@ int main(int argc, char* argv[]) tryReloadTreeFromDisk(edgeBus, localTreePath); }); - if (!session.connectToCore(coreHost, corePort)) { - qWarning() << "cannot reach core; start softbus_daemon with edgeIngress.enabled and matching port," - << "or set coreHost/corePort in edge_manifest.json"; - return 3; - } - if (!session.sendHello(edgeId, authToken, 0)) { - qWarning() << "hello send failed"; - return 4; - } - - if (edgeDiscoveryEnabled) { - (void)discovery.start(); - } + QObject::connect(&session, &softbus::message_bus::ingress::EdgeSession::receivedDebugTunnel, &app, + [&](const QJsonObject& envelope) { + qInfo() << "DebugTunnel from host" << envelope; + }); OpsUiHttp ops; if (opsPort > 0) { @@ -171,42 +420,62 @@ int main(int argc, char* argv[]) qWarning() << "propose send failed"; } }, - [&discovery]() { return discovery.recentDiscoveriesJson(); })) { + [&discovery]() { return discovery.recentDiscoveriesJson(); }, + [&](const QJsonObject& dbg) { return session.sendDebugToCore(dbg); }, + [&session]() { return session.recentDebugFromHost(); }, + opsListenAll)) { qWarning() << "ops ui listen failed"; } else { - qInfo() << "ops UI http://127.0.0.1:" << opsPort - << " GET / GET /ui GET /discovered POST /propose POST /register_discovered"; + qInfo() << "ops UI" + << (opsListenAll ? QStringLiteral("http://0.0.0.0:") : QStringLiteral("http://127.0.0.1:")) + << opsPort << "GET / GET /ui GET /discovered GET /debug GET /debug_host POST /propose" + " POST /register_discovered POST /debug_edge"; } } - FakeDriver fake; - if (fakeIntervalMs > 0) { - fake.start(fakeIntervalMs, [&]() { - const std::uint32_t epHash = static_cast(qHash(endpoint)); - const std::uint64_t routing = (static_cast(epHash) << 32) - | static_cast(deviceId); - const QByteArray inner = softbus::message_bus::ingress::edge_uplink::encodeDataRawBusPayload( - deviceId, - epHash, - static_cast(protocol), - 0, - routing, - 0, - QDateTime::currentMSecsSinceEpoch(), - 1, - QStringLiteral("edge:fake"), - QByteArray(1, '\x01')); - if (inner.isEmpty()) { - qWarning() << "encode data failed"; - return; - } - if (!session.sendDataRawBusInner(inner)) { - qWarning() << "send data failed"; - } - }); + discovery.setStandaloneLocalOnly(true); + session.configureUplink( + coreHost, + corePort, + edgeId, + authToken, + [&]() { return std::max(coreTreeRevision, edgeBus.treeModel().treeRevision()); }); + QObject::connect(&session, &softbus::message_bus::ingress::EdgeSession::uplinkReady, &app, [&]() { + discovery.setStandaloneLocalOnly(false); + discovery.requestFlush(); + }); + QObject::connect(&session, &softbus::message_bus::ingress::EdgeSession::uplinkLost, &app, [&]() { + discovery.setStandaloneLocalOnly(true); + }); + + if (edgeDiscoveryEnabled) { + if (!discovery.start()) { + qWarning() << "EdgeDiscoveryCoordinator: udev discovery did not start"; + } } - QObject::connect(&app, &QCoreApplication::aboutToQuit, &discovery, [&discovery]() { discovery.stop(); }); + session.beginUplinkLoop(); + + QTimer heartbeatTimer; + if (heartbeatIntervalMs > 0) { + heartbeatTimer.setInterval(std::clamp(heartbeatIntervalMs, 100, 24 * 3600 * 1000)); + heartbeatTimer.setTimerType(Qt::CoarseTimer); + QObject::connect(&heartbeatTimer, &QTimer::timeout, &app, [&session, &edgeBus, edgeId, &coreTreeRevision]() { + if (!session.isCoreTcpConnected()) { + return; + } + const QJsonObject body = makeEdgeHeartbeatPayload(edgeId, coreTreeRevision, edgeBus); + if (!session.sendDebugToCore(body)) { + qWarning() << "edge heartbeat (DEBUG_TUNNEL) send failed"; + } + }); + heartbeatTimer.start(); + } + + QObject::connect(&app, &QCoreApplication::aboutToQuit, &app, [&discovery, &session]() { + session.shutdownUplink(); + discovery.stop(); + }); (void)tcpUplink; return app.exec(); diff --git a/src/message_bus/ingress/EdgeSession.cpp b/src/message_bus/ingress/EdgeSession.cpp index e14dc04..ef7d7b5 100644 --- a/src/message_bus/ingress/EdgeSession.cpp +++ b/src/message_bus/ingress/EdgeSession.cpp @@ -1,7 +1,9 @@ #include "message_bus/ingress/EdgeSession.h" +#include #include #include +#include #include "message_bus/ingress/EdgeUplinkWire.h" @@ -18,6 +20,123 @@ EdgeSession::EdgeSession(QObject* parent) connect(&m_sock, &QTcpSocket::readyRead, this, &EdgeSession::onSocketReadyRead); } +void EdgeSession::configureUplink(const QString& host, + int port, + const QString& edgeId, + const QString& authToken, + std::function currentTreeRevision) +{ + m_uplinkHost = host; + m_uplinkPort = port; + m_uplinkEdgeId = edgeId; + m_uplinkAuthToken = authToken; + m_uplinkTreeRevision = std::move(currentTreeRevision); + + if (!m_reconnectTimer) { + m_reconnectTimer = new QTimer(this); + m_reconnectTimer->setSingleShot(true); + connect(m_reconnectTimer, &QTimer::timeout, this, &EdgeSession::kickConnectInternal); + } + + if (!m_uplinkHooksInstalled) { + m_uplinkHooksInstalled = true; + connect(&m_sock, &QTcpSocket::connected, this, &EdgeSession::onTcpConnected); + connect(&m_sock, &QTcpSocket::disconnected, this, &EdgeSession::onTcpDisconnected); + connect(&m_sock, &QAbstractSocket::errorOccurred, this, &EdgeSession::onTcpError); + } +} + +void EdgeSession::beginUplinkLoop() +{ + m_uplinkShutdown = false; + kickConnectInternal(); +} + +void EdgeSession::shutdownUplink() +{ + m_uplinkShutdown = true; + if (m_reconnectTimer) { + m_reconnectTimer->stop(); + } + m_sock.abort(); +} + +bool EdgeSession::isCoreTcpConnected() const +{ + return m_sock.state() == QAbstractSocket::ConnectedState; +} + +void EdgeSession::kickConnectInternal() +{ + if (m_uplinkShutdown || m_uplinkHost.isEmpty() || m_uplinkPort <= 0) { + return; + } + if (m_sock.state() == QAbstractSocket::ConnectedState + || m_sock.state() == QAbstractSocket::ConnectingState) { + return; + } + if (m_sock.state() != QAbstractSocket::UnconnectedState) { + m_sock.abort(); + } + qInfo() << "EdgeSession: connecting to core" << m_uplinkHost << m_uplinkPort; + m_sock.connectToHost(m_uplinkHost, static_cast(m_uplinkPort)); +} + +void EdgeSession::planReconnect() +{ + if (m_uplinkShutdown || !m_reconnectTimer) { + return; + } + if (!m_reconnectTimer->isActive()) { + qInfo() << "EdgeSession: scheduling core reconnect in 3s"; + m_reconnectTimer->start(3000); + } +} + +void EdgeSession::onTcpConnected() +{ + if (m_uplinkShutdown) { + return; + } + if (m_reconnectTimer) { + m_reconnectTimer->stop(); + } + const qint64 tr = m_uplinkTreeRevision ? m_uplinkTreeRevision() : 0; + if (!sendHello(m_uplinkEdgeId, m_uplinkAuthToken, tr)) { + qWarning() << "EdgeSession: HELLO write failed"; + emit uplinkLost(); + m_sock.abort(); + planReconnect(); + return; + } + qInfo() << "EdgeSession: TCP up, HELLO sent currentTreeRevision" << tr; + flushPendingSbupFrames(); + emit uplinkReady(); +} + +void EdgeSession::onTcpDisconnected() +{ + if (m_uplinkShutdown) { + return; + } + qWarning() << "EdgeSession: disconnected from core"; + emit uplinkLost(); + planReconnect(); +} + +void EdgeSession::onTcpError(QAbstractSocket::SocketError) +{ + if (m_uplinkShutdown) { + return; + } + qWarning() << "EdgeSession: socket error" << m_sock.errorString(); + emit uplinkLost(); + if (m_sock.state() != QAbstractSocket::UnconnectedState) { + m_sock.abort(); + } + planReconnect(); +} + bool EdgeSession::connectToCore(const QString& host, int port) { m_sock.connectToHost(host, static_cast(port)); @@ -35,6 +154,9 @@ void EdgeSession::disconnectCore() bool EdgeSession::sendHello(const QString& edgeId, const QString& authToken, qint64 currentTreeRevision) { + if (m_sock.state() != QAbstractSocket::ConnectedState) { + return false; + } QJsonObject o; o.insert(QStringLiteral("edgeId"), edgeId); if (!authToken.isEmpty()) { @@ -48,20 +170,89 @@ bool EdgeSession::sendHello(const QString& edgeId, const QString& authToken, qin bool EdgeSession::sendDataRawBusInner(const QByteArray& innerPayload) { + if (m_sock.state() != QAbstractSocket::ConnectedState) { + return false; + } const QByteArray frame = encodeOuterFrame(FrameKind::DataRawBus, innerPayload); return m_sock.write(frame) == frame.size(); } bool EdgeSession::sendTreePropose(const QJsonObject& body) { + if (m_sock.state() != QAbstractSocket::ConnectedState) { + return false; + } const QByteArray pl = QJsonDocument(body).toJson(QJsonDocument::Compact); const QByteArray frame = encodeOuterFrame(FrameKind::TreePropose, pl); return m_sock.write(frame) == frame.size(); } +bool EdgeSession::sendDebugToCore(const QJsonObject& body) +{ + QJsonObject envelope; + envelope.insert(QStringLiteral("v"), 1); + envelope.insert(QStringLiteral("from"), QStringLiteral("edge")); + envelope.insert(QStringLiteral("tsMs"), QDateTime::currentMSecsSinceEpoch()); + envelope.insert(QStringLiteral("body"), body); + const QByteArray pl = QJsonDocument(envelope).toJson(QJsonDocument::Compact); + const QByteArray frame = encodeOuterFrame(FrameKind::DebugTunnel, pl); + if (m_sock.state() != QAbstractSocket::ConnectedState) { + return false; + } + return m_sock.write(frame) == frame.size(); +} + +void EdgeSession::flushPendingSbupFrames() +{ + if (m_sock.state() != QAbstractSocket::ConnectedState) { + return; + } + while (!m_pendingSbupFrames.isEmpty()) { + const QByteArray frame = m_pendingSbupFrames.takeFirst(); + if (m_sock.write(frame) != frame.size()) { + m_pendingSbupFrames.prepend(frame); + break; + } + } +} + bool EdgeSession::writeSbupFrame(const QByteArray& frame) { - return m_sock.write(frame) == frame.size(); + if (frame.isEmpty()) { + return false; + } + if (m_sock.state() != QAbstractSocket::ConnectedState) { + if (m_pendingSbupFrames.size() < kMaxPendingSbupFrames) { + m_pendingSbupFrames.push_back(frame); + return true; + } + return false; + } + if (m_sock.write(frame) != frame.size()) { + if (m_pendingSbupFrames.size() < kMaxPendingSbupFrames) { + m_pendingSbupFrames.push_back(frame); + } + return false; + } + flushPendingSbupFrames(); + return true; +} + +void EdgeSession::appendDebugFromHost(const QJsonObject& envelope) +{ + m_debugFromHostRing.push_back(envelope); + if (m_debugFromHostRing.size() > kMaxDebugRing) { + m_debugFromHostRing.remove(0, m_debugFromHostRing.size() - kMaxDebugRing); + } +} + +QJsonArray EdgeSession::recentDebugFromHost() const +{ + QJsonArray arr; + for (const QJsonObject& o : m_debugFromHostRing) { + arr.append(o); + } + return arr; } void EdgeSession::onSocketReadyRead() @@ -84,6 +275,14 @@ void EdgeSession::onSocketReadyRead() if (pe.error == QJsonParseError::NoError && doc.isObject()) { emit receivedTreeFull(doc.object()); } + } else if (popped->kind == FrameKind::DebugTunnel) { + QJsonParseError pe; + const QJsonDocument doc = QJsonDocument::fromJson(popped->payload, &pe); + if (pe.error == QJsonParseError::NoError && doc.isObject()) { + const QJsonObject o = doc.object(); + appendDebugFromHost(o); + emit receivedDebugTunnel(o); + } } } } diff --git a/src/message_bus/ingress/EdgeSession.h b/src/message_bus/ingress/EdgeSession.h index 8e6793b..7cf589f 100644 --- a/src/message_bus/ingress/EdgeSession.h +++ b/src/message_bus/ingress/EdgeSession.h @@ -1,9 +1,17 @@ #pragma once +#include + +#include #include +#include #include #include +#include #include +#include + +class QTimer; namespace softbus::message_bus::ingress { @@ -17,25 +25,69 @@ public: bool connectToCore(const QString& host, int port); void disconnectCore(); + /** + * 配置核心 TCP 与 HELLO 参数，并启用自动重连（核心晚于边缘启动时仍可连上）。 + * 调用 `beginUplinkLoop()` 后开始尝试连接；断线或拒绝连接后约 3s 重试。 + */ + void configureUplink(const QString& host, + int port, + const QString& edgeId, + const QString& authToken, + std::function currentTreeRevision); + void beginUplinkLoop(); + void shutdownUplink(); + bool isCoreTcpConnected() const; + bool sendHello(const QString& edgeId, const QString& authToken, qint64 currentTreeRevision); bool sendDataRawBusInner(const QByteArray& innerPayload); bool sendTreePropose(const QJsonObject& body); + /** 经 SBUP `DebugTunnel` 向核心发送 JSON（外层载荷为 UTF-8 JSON）。 */ + bool sendDebugToCore(const QJsonObject& body); /** 已带 SBUP 外层的任意帧（如完整 `DATA_RAWBUS`）。 */ bool writeSbupFrame(const QByteArray& frame); void setMaxPayloadBytes(int m) { m_maxPayload = m; } + /** 最近从主机收到的 `DebugTunnel` 记录（环形，用于 Ops `/debug_host`）。 */ + QJsonArray recentDebugFromHost() const; + signals: void receivedTreeResult(const QJsonObject& obj); void receivedTreeFull(const QJsonObject& obj); + void receivedDebugTunnel(const QJsonObject& obj); + /** TCP 已连上且 HELLO 帧已写入套接字（之后由核心下发 TREE_RESULT 等）。 */ + void uplinkReady(); + /** 与核心 TCP 断开或连接失败进入待机重连。 */ + void uplinkLost(); private slots: void onSocketReadyRead(); + void onTcpConnected(); + void onTcpDisconnected(); + void onTcpError(QAbstractSocket::SocketError socketError); private: + void appendDebugFromHost(const QJsonObject& envelope); + void kickConnectInternal(); + void planReconnect(); + void flushPendingSbupFrames(); + QTcpSocket m_sock; QByteArray m_inbuf; int m_maxPayload{4 * 1024 * 1024}; + QVector m_debugFromHostRing; + static constexpr int kMaxDebugRing = 64; + + QString m_uplinkHost; + int m_uplinkPort{0}; + QString m_uplinkEdgeId; + QString m_uplinkAuthToken; + std::function m_uplinkTreeRevision; + QTimer* m_reconnectTimer{nullptr}; + bool m_uplinkHooksInstalled{false}; + bool m_uplinkShutdown{false}; + QVector m_pendingSbupFrames; + static constexpr int kMaxPendingSbupFrames = 512; }; } // namespace softbus::message_bus::ingress diff --git a/src/message_bus/ingress/EdgeTcpUplinkPort.cpp b/src/message_bus/ingress/EdgeTcpUplinkPort.cpp index 459402f..7fc38d7 100644 --- a/src/message_bus/ingress/EdgeTcpUplinkPort.cpp +++ b/src/message_bus/ingress/EdgeTcpUplinkPort.cpp @@ -6,6 +6,7 @@ #include #include "message_bus/ingress/EdgeUplinkWire.h" +#include "utils/AgentDebugNdjson5448.h" namespace softbus::message_bus::ingress { @@ -42,7 +43,19 @@ bool EdgeTcpUplinkPort::push(softbus::core::models::RawBusMessage ctx) } const QByteArray outer = edge_uplink::encodeOuterFrame(edge_uplink::FrameKind::DataRawBus, inner); - return m_writer(outer); + const bool ok = m_writer(outer); + // #region agent log + if (ok || busPayload.size() >= 4) { + QJsonObject d; + d.insert(QStringLiteral("ok"), ok); + d.insert(QStringLiteral("deviceId"), ctx.deviceId); + d.insert(QStringLiteral("protocol"), static_cast(ctx.protocol)); + d.insert(QStringLiteral("busPayloadLen"), busPayload.size()); + d.insert(QStringLiteral("outerLen"), outer.size()); + agent_debug_5448::log("H-uplink", "EdgeTcpUplinkPort.cpp:push", "encode_data_rawbus", d); + } + // #endregion + return ok; } void EdgeTcpUplinkPort::reportError(const QString& endpoint, const QString& error) diff --git a/src/message_bus/ingress/EdgeUplinkWire.h b/src/message_bus/ingress/EdgeUplinkWire.h index 734a1a5..59658a7 100644 --- a/src/message_bus/ingress/EdgeUplinkWire.h +++ b/src/message_bus/ingress/EdgeUplinkWire.h @@ -27,6 +27,8 @@ enum class FrameKind : std::uint16_t TreePropose = 6, TreeResult = 7, Report = 8, + /** 双向 UTF-8 JSON 调试/监控（主机↔边缘，需已 HELLO）。见 softbus_daemon/docs/EDGE_UPLINK_PROTOCOL_v1.md §10。 */ + DebugTunnel = 9, }; inline constexpr std::size_t kOuterHeaderBytes = 16; diff --git a/src/ops_ui_http.cpp b/src/ops_ui_http.cpp index 7e03bb3..17d3b39 100644 --- a/src/ops_ui_http.cpp +++ b/src/ops_ui_http.cpp @@ -1,5 +1,6 @@ #include "ops_ui_http.hpp" +#include #include #include #include @@ -79,7 +80,7 @@ QByteArray buildUiPage(const QString& edgeId, const QJsonArray& discovered) "" "" "" - "

本地树 JSON · 发现 JSON · POST /propose 仍可提交完整 JSON。

" + "

本地树 JSON · 发现 JSON · 调试通道 · POST /propose 仍可提交完整 JSON。

" "") .arg(htmlEscape(edgeId), options); return html.toUtf8(); @@ -130,16 +131,23 @@ bool OpsUiHttp::start(quint16 port, const QString& edgeId, const std::function& getBaseTreeRevision, const std::function& onPropose, - const std::function& getDiscovered) + const std::function& getDiscovered, + const std::function& sendDebugToCore, + const std::function& getDebugFromHostInbox, + bool listenOnAllInterfaces) { m_treePath = localTreePath; m_edgeId = edgeId; m_getBaseTreeRevision = getBaseTreeRevision; m_onPropose = onPropose; m_getDiscovered = getDiscovered; + m_sendDebugToCore = sendDebugToCore; + m_getDebugFromHostInbox = getDebugFromHostInbox; m_tcp = new QTcpServer(this); connect(m_tcp, &QTcpServer::newConnection, this, &OpsUiHttp::onNewConnection); - if (!m_tcp->listen(QHostAddress::LocalHost, port)) { + const QHostAddress bindAddr = listenOnAllInterfaces ? QHostAddress::Any : QHostAddress::LocalHost; + if (!m_tcp->listen(bindAddr, port)) { + qWarning() << "OpsUiHttp: listen failed" << bindAddr << port << m_tcp->errorString(); return false; } return true; @@ -235,6 +243,50 @@ void OpsUiHttp::handleSocket(QTcpSocket* sock, const QByteArray& data) return; } + if (method == "GET" && path == "/debug_host") { + if (!m_getDebugFromHostInbox) { + const QJsonObject err{{QStringLiteral("error"), QStringLiteral("debug inbox not wired")}}; + writeHttp(sock, 503, QJsonDocument(err).toJson(QJsonDocument::Compact), "application/json; charset=utf-8"); + return; + } + const QJsonArray items = m_getDebugFromHostInbox(); + const QByteArray out = + QJsonDocument(QJsonObject{{QStringLiteral("items"), items}}).toJson(QJsonDocument::Indented); + writeHttp(sock, 200, out, "application/json; charset=utf-8"); + return; + } + + if (method == "POST" && path == "/debug_edge") { + if (!m_sendDebugToCore) { + const QJsonObject err{{QStringLiteral("error"), QStringLiteral("debug uplink not wired")}}; + writeHttp(sock, 503, QJsonDocument(err).toJson(QJsonDocument::Compact), "application/json; charset=utf-8"); + return; + } + QJsonParseError pe; + const QJsonDocument doc = QJsonDocument::fromJson(body, &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + writeHttp(sock, 400, QByteArrayLiteral("invalid json (expect object)"), "text/plain; charset=utf-8"); + return; + } + const bool ok = m_sendDebugToCore(doc.object()); + const QJsonObject resp{{QStringLiteral("ok"), ok}}; + writeHttp(sock, ok ? 200 : 502, QJsonDocument(resp).toJson(QJsonDocument::Compact), "application/json; charset=utf-8"); + return; + } + + if (method == "GET" && path == "/debug") { + const QByteArray html = + QByteArrayLiteral("调试通道" + "

主机 ↔ 边缘 DebugTunnel

" + "
    " + "
  • GET /debug_host — 最近从主机经 SBUP 下发的调试 JSON（环形缓冲）
    • " + "
  • POST /debug_edge — body 为 JSON 对象，经 SBUP 发往主机（需已 HELLO）
    • " + "
" + "

打开 debug_host JSON · 配置页

"); + writeHttp(sock, 200, html, "text/html; charset=utf-8"); + return; + } + if (method == "GET" && path == "/ui") { QJsonArray arr; if (m_getDiscovered) { diff --git a/src/ops_ui_http.hpp b/src/ops_ui_http.hpp index 2187126..770b3c1 100644 --- a/src/ops_ui_http.hpp +++ b/src/ops_ui_http.hpp @@ -22,7 +22,10 @@ public: const QString& edgeId, const std::function& getBaseTreeRevision, const std::function& onPropose, - const std::function& getDiscovered = {}); + const std::function& getDiscovered = {}, + const std::function& sendDebugToCore = {}, + const std::function& getDebugFromHostInbox = {}, + bool listenOnAllInterfaces = false); private slots: void onNewConnection(); @@ -36,4 +39,6 @@ private: std::function m_getBaseTreeRevision; std::function m_onPropose; std::function m_getDiscovered; + std::function m_sendDebugToCore; + std::function m_getDebugFromHostInbox; }; diff --git a/src/pipeline/EdgeStreamPipeline.cpp b/src/pipeline/EdgeStreamPipeline.cpp new file mode 100644 index 0000000..5f4170e --- /dev/null +++ b/src/pipeline/EdgeStreamPipeline.cpp @@ -0,0 +1,399 @@ +#include "pipeline/EdgeStreamPipeline.h" + +#include "device_bus/tree/DeviceTreeModel.h" + +#include +#include +#include +#include + +#include +#include + +#include "core/memory/LockFreeQueue.h" +#include "core/plugin_system/IMapperDependencies.h" +#include "utils/logs/logging.h" +#include "utils/AgentDebugNdjson5448.h" + +namespace +{ +QString edgePayloadHexPreview(const softbus::core::models::RawBusMessage& ctx, int maxBytes) +{ + if (!ctx.payload.valid()) { + return QStringLiteral("(invalid)"); + } + const std::uint8_t* p = ctx.payload.bytes(); + const std::size_t n = ctx.payload.length; + if (!p || n == 0) { + return QStringLiteral("(empty)"); + } + QString s; + const std::size_t lim = (std::min)(n, static_cast(maxBytes)); + for (std::size_t i = 0; i < lim; ++i) { + s += QStringLiteral("%1 ").arg(static_cast(p[i]), 2, 16, QLatin1Char('0')); + } + if (lim < n) { + s += QStringLiteral("…"); + } + return s.trimmed(); +} +} // namespace + +namespace softbus_edge +{ + +struct EdgeFramerEndpointState +{ + softbus::core::memory::LockFreeQueue chunkQ; + std::atomic running{true}; + std::thread th; + std::uint32_t endpointHash{0}; + EdgeStreamPipeline* pipeline{nullptr}; + + explicit EdgeFramerEndpointState(std::size_t cap, std::uint32_t ep, EdgeStreamPipeline* pl) + : chunkQ(cap) + , endpointHash(ep) + , pipeline(pl) + { + } +}; + +static void edgeFramerThreadMain(std::shared_ptr st); + +EdgeStreamPipeline::EdgeStreamPipeline(Config cfg) + : m_cfg(std::move(cfg)) + , m_treeModel(m_cfg.treeModel) +{ + if (!m_cfg.pool) { + m_lastError = QStringLiteral("memory_pool_null"); + return; + } + m_pluginManager = std::make_shared(); + m_pluginHost = std::make_unique(m_pluginManager); + m_passthroughFramer = + std::make_unique(m_cfg.pool); + + for (const QString& path : m_cfg.protocolPluginPaths) { + if (path.trimmed().isEmpty()) { + continue; + } + QString err; + if (!m_pluginHost->loadPlugin(path, &err)) { + m_lastError = QStringLiteral("load_plugin_failed:%1:%2").arg(path, err); + return; + } + } + if (m_pluginManager && m_cfg.pool) { + const softbus::core::models::ProtocolType types[] = { + softbus::core::models::ProtocolType::MODBUS_RTU, + softbus::core::models::ProtocolType::MODBUS_ASCII, + softbus::core::models::ProtocolType::MODBUS_TCP, + }; + for (softbus::core::models::ProtocolType t : types) { + if (auto fr = m_pluginManager->selectProtocolFramerPlugin(t)) { + fr->bindMemoryPool(m_cfg.pool); + } + } + } + m_ok = true; +} + +EdgeStreamPipeline::~EdgeStreamPipeline() +{ + shutdownFramerWorkers(); +} + +void EdgeStreamPipeline::setFramedSink(std::function sink) +{ + m_framedSink = std::move(sink); +} + +void EdgeStreamPipeline::shutdownFramerWorkers() +{ + std::unordered_map> workers; + { + std::lock_guard lk(m_framerWorkerMutex); + workers.swap(m_framerWorkers); + } + for (auto& kv : workers) { + if (kv.second) { + kv.second->running.store(false, std::memory_order_release); + } + } + for (auto& kv : workers) { + if (kv.second && kv.second->th.joinable()) { + kv.second->th.join(); + } + } +} + +std::shared_ptr EdgeStreamPipeline::ensureFramerWorker(std::uint32_t endpointHash) +{ + std::lock_guard lk(m_framerWorkerMutex); + const auto it = m_framerWorkers.find(endpointHash); + if (it != m_framerWorkers.end()) { + return it->second; + } + const std::size_t cap = m_cfg.edgeSerialChunkQueueCapacity > 0 ? m_cfg.edgeSerialChunkQueueCapacity : 4096; + auto st = std::make_shared(cap, endpointHash, this); + st->th = std::thread(edgeFramerThreadMain, st); + m_framerWorkers.emplace(endpointHash, st); + return st; +} + +bool EdgeStreamPipeline::tryEnqueueStreamChunk(softbus::core::models::RawBusMessage chunk) +{ + if (!m_ok) { + return false; + } + if (!m_cfg.edgeUseEndpointFramerThreads) { + processFramerIngressFromWorker(std::move(chunk)); + return true; + } + const std::shared_ptr st = ensureFramerWorker(chunk.endpointHash); + if (!st) { + return false; + } + return st->chunkQ.push(std::move(chunk)); +} + +void EdgeStreamPipeline::processFramerIngressFromWorker(softbus::core::models::RawBusMessage&& chunkIn) +{ + if (!m_ok) { + return; + } + const qint64 inChunkLen = + chunkIn.payload.valid() ? static_cast(chunkIn.payload.length) : qint64{-1}; + QString hexPreview; + if (chunkIn.payload.valid() && chunkIn.payload.length > 0) { + hexPreview = edgePayloadHexPreview(chunkIn, 48); + } + std::vector framedOut; + dispatchFramer(chunkIn, framedOut); + // #region agent log + { + QJsonObject d; + d.insert(QStringLiteral("inChunkLen"), inChunkLen); + d.insert(QStringLiteral("framesOut"), static_cast(framedOut.size())); + d.insert(QStringLiteral("protocol"), static_cast(chunkIn.protocol)); + d.insert(QStringLiteral("deviceId"), chunkIn.deviceId); + d.insert(QStringLiteral("endpointHash"), QString::number(chunkIn.endpointHash, 16)); + if (!chunkIn.traceId.isEmpty()) { + d.insert(QStringLiteral("traceId"), chunkIn.traceId); + } + if (!hexPreview.isEmpty()) { + d.insert(QStringLiteral("hexPreview"), hexPreview); + } + if (m_cfg.pool) { + d.insert(QStringLiteral("poolInUse"), static_cast(m_cfg.pool->inUseBlocks())); + d.insert(QStringLiteral("poolTotal"), static_cast(m_cfg.pool->totalBlocks())); + } + agent_debug_5448::log("H-framer", "EdgeStreamPipeline.cpp:processFramer", "after_dispatch", d); + } + // #endregion + for (auto& framed : framedOut) { + auto doms = runParserMapper(framed); + attachEdgeContext(framed, doms); + // #region agent log + { + QJsonObject d; + d.insert(QStringLiteral("domCount"), static_cast(doms.size())); + d.insert(QStringLiteral("frameLen"), + framed.payload.valid() ? static_cast(framed.payload.length) : qint64{-1}); + d.insert(QStringLiteral("frameHex"), edgePayloadHexPreview(framed, 32)); + agent_debug_5448::log("H-parse", "EdgeStreamPipeline.cpp:runParserMapper", "dom_result", d); + } + // #endregion + for (const auto& d : doms) { + LOG_INFO() << "EdgeStreamPipeline: parsed DOM messageId=" << d.messageId << "pointId=" << d.pointId + << "metadataId=" << d.metadataId; + } + if (m_framedSink) { + const std::uint32_t epH = framed.endpointHash; + const qint64 frameLen = + framed.payload.valid() ? static_cast(framed.payload.length) : qint64{-1}; + const bool pushed = m_framedSink(std::move(framed)); + // #region agent log + { + QJsonObject d; + d.insert(QStringLiteral("pushed"), pushed); + d.insert(QStringLiteral("frameLen"), frameLen); + d.insert(QStringLiteral("endpointHash"), QString::number(epH, 16)); + agent_debug_5448::log("H-uplink", "EdgeStreamPipeline.cpp:framedSink", "data_rawbus_push", d); + } + // #endregion + if (!pushed) { + LOG_WARNING() << "EdgeStreamPipeline: framedSink push failed endpointHash=0x" + << QString::number(static_cast(epH), 16); + } + } else { + LOG_WARNING() << "EdgeStreamPipeline: framedSink not set, dropping framed frame"; + } + } +} + +static void edgeFramerThreadMain(std::shared_ptr st) +{ + while (st->running.load(std::memory_order_acquire)) { + softbus::core::models::RawBusMessage ch; + if (!st->chunkQ.pop(ch)) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + } + if (st->pipeline) { + st->pipeline->processFramerIngressFromWorker(std::move(ch)); + } + } +} + +void EdgeStreamPipeline::dispatchFramer(softbus::core::models::RawBusMessage& chunkIn, + std::vector& completeFramesOut) +{ + if (m_pluginManager) { + auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocol); + if (plugin) { + plugin->bindMemoryPool(m_cfg.pool); + const PluginSessionKey key{chunkIn.endpointHash, plugin->pluginId()}; + std::lock_guard lk(m_framerSessionMutex); + auto it = m_framerSessions.find(key); + if (it == m_framerSessions.end() || !it.value()) { + auto s = plugin->createFramerSession(chunkIn.endpointHash, QJsonObject{}); + m_framerSessions.insert( + key, + s ? std::shared_ptr(std::move(s)) : nullptr); + it = m_framerSessions.find(key); + } + if (it != m_framerSessions.end() && it.value()) { + it.value()->feedChunk(chunkIn, completeFramesOut); + return; + } + } + } + if (m_passthroughFramer) { + m_passthroughFramer->feed(chunkIn, completeFramesOut); + } +} + +std::vector EdgeStreamPipeline::runParserMapper( + const softbus::core::models::RawBusMessage& ctx) +{ + std::vector out; + if (!m_pluginManager) { + return out; + } + auto parserPlugin = m_pluginManager->selectProtocolPlugin(ctx.protocol); + if (!parserPlugin) { + if (ctx.protocol != softbus::core::models::ProtocolType::UNKNOWN) { + LOG_WARNING() << "EdgeStreamPipeline: no parser plugin for protocol" << static_cast(ctx.protocol) + << "endpointHash=0x" << QString::number(ctx.endpointHash, 16) << "payloadLen=" + << (ctx.payload.valid() ? static_cast(ctx.payload.length) : qint64{-1}); + } + return out; + } + const PluginSessionKey parserKey{ctx.endpointHash, parserPlugin->pluginId()}; + std::shared_ptr parserSession; + { + std::lock_guard lk(m_parserSessionMutex); + auto it = m_parserSessions.find(parserKey); + if (it == m_parserSessions.end() || !it.value()) { + auto s = parserPlugin->createSession(); + m_parserSessions.insert( + parserKey, + s ? std::shared_ptr(std::move(s)) : nullptr); + it = m_parserSessions.find(parserKey); + } + if (it == m_parserSessions.end() || !it.value()) { + LOG_WARNING() << "EdgeStreamPipeline: createParserSession failed for" << parserPlugin->pluginId(); + return out; + } + parserSession = it.value(); + } + softbus::message_bus::pipeline::protocol::ProtocolEnvelope envelope; + if (!parserSession->parse(ctx, envelope)) { + LOG_WARNING() << "EdgeStreamPipeline: parse failed protocol" << static_cast(ctx.protocol) << "payloadLen=" + << (ctx.payload.valid() ? static_cast(ctx.payload.length) : qint64{-1}) + << "hex=" << edgePayloadHexPreview(ctx, 24); + return out; + } + + auto mapperPlugin = m_pluginManager->selectProtocolMapperPlugin(ctx.protocol); + if (!mapperPlugin) { + LOG_WARNING() << "EdgeStreamPipeline: no mapper plugin for protocol" << static_cast(ctx.protocol); + return out; + } + const PluginSessionKey mapperKey{ctx.endpointHash, mapperPlugin->pluginId()}; + std::shared_ptr mapperSession; + { + std::lock_guard lk(m_mapperSessionMutex); + auto it = m_mapperSessions.find(mapperKey); + if (it == m_mapperSessions.end() || !it.value()) { + softbus::core::plugin_system::MapperSessionContext context; + context.mappingLookup = nullptr; + context.discoverySink = nullptr; + context.sniffingEnabled = []() { return true; }; + auto s = mapperPlugin->createMapperSession(context); + m_mapperSessions.insert( + mapperKey, + s ? std::shared_ptr(std::move(s)) : nullptr); + it = m_mapperSessions.find(mapperKey); + } + if (it == m_mapperSessions.end() || !it.value()) { + LOG_WARNING() << "EdgeStreamPipeline: createMapperSession failed for" << mapperPlugin->pluginId(); + return out; + } + mapperSession = it.value(); + } + return mapperSession->map(ctx, envelope); +} + +void EdgeStreamPipeline::attachEdgeContext(const softbus::core::models::RawBusMessage& raw, + std::vector& msgs) const +{ + if (msgs.empty()) { + return; + } + const DeviceTreeNode* node = nullptr; + if (m_treeModel && raw.deviceId >= 0) { + for (const auto& n : m_treeModel->nodes()) { + const int did = n.params.value(QStringLiteral("deviceId")).toInt(-999999); + if (did == raw.deviceId) { + node = &n; + break; + } + } + } + for (auto& msg : msgs) { + msg.attributes.insert(QStringLiteral("endpointHash"), static_cast(raw.endpointHash)); + if (raw.deviceId >= 0) { + msg.attributes.insert(QStringLiteral("runtimeDeviceId"), raw.deviceId); + } + if (node) { + msg.attributes.insert(QStringLiteral("deviceNodeId"), node->id); + msg.attributes.insert(QStringLiteral("deviceEndpoint"), node->endpoint); + const QJsonValue av = node->params.value(QStringLiteral("assetId")); + if (av.isString() && !av.toString().isEmpty()) { + msg.attributes.insert(QStringLiteral("assetId"), av.toString()); + if (msg.objectType.isEmpty() || msg.objectType == QLatin1String("system")) { + msg.objectType = QStringLiteral("asset"); + msg.objectRef = av.toString(); + } + } + } + if (!msg.pointId.isEmpty()) { + msg.attributes.insert(QStringLiteral("pointId"), msg.pointId); + } + if (!msg.metadataId.isEmpty()) { + msg.attributes.insert(QStringLiteral("metadataId"), msg.metadataId); + } + if (msg.objectType.isEmpty()) { + msg.objectType = QStringLiteral("device"); + } + if (msg.objectRef.isEmpty()) { + msg.objectRef = raw.deviceId >= 0 ? QString::number(raw.deviceId) : QStringLiteral("default"); + } + msg.attributes.insert(QStringLiteral("objectType"), msg.objectType); + msg.attributes.insert(QStringLiteral("objectRef"), msg.objectRef); + } +} + +} // namespace softbus_edge diff --git a/src/pipeline/EdgeStreamPipeline.h b/src/pipeline/EdgeStreamPipeline.h new file mode 100644 index 0000000..c2ff918 --- /dev/null +++ b/src/pipeline/EdgeStreamPipeline.h @@ -0,0 +1,119 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "core/memory/MemoryPool.h" +#include "core/models/DOMMessage.h" +#include "core/models/RawBusMessage.h" +#include "core/plugin_system/IProtocolFramerPlugin.h" +#include "core/plugin_system/IProtocolMapperPlugin.h" +#include "core/plugin_system/IProtocolPlugin.h" +#include "core/plugin_system/PluginHost.h" +#include "core/plugin_system/PluginManager.h" +#include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h" + +class DeviceTreeModel; + +namespace softbus_edge +{ + +struct EdgeFramerEndpointState; + +struct PluginSessionKey +{ + std::uint32_t endpointHash{0}; + QString pluginId; + bool operator==(const PluginSessionKey& o) const + { + return endpointHash == o.endpointHash && pluginId == o.pluginId; + } +}; + +/** + * 与 softbus_daemon Pipeline 的 dispatchFramer + stage2Parser（解析+mapper）语义对齐；用于边缘串口流。 + * 默认按 `endpointHash` 维护与 daemon `EndpointFramingState` 相同的 **SPSC LockFreeQueue + 独立分帧线程**， + * 串口回调仅入队；线程内完成 framer→parser→mapper 并经 `setFramedSink` 上行。 + */ +class EdgeStreamPipeline final +{ +public: + struct Config + { + std::shared_ptr pool; + QStringList protocolPluginPaths; + /** 可选：用于在 DOM 上附加 deviceNodeId / assetId 等（轻量版 attachPesmRuntimeContext）。 */ + DeviceTreeModel* treeModel{nullptr}; + /** 与 daemon `EndpointFramingState::chunkQ` 容量一致的有界队列深度（SPSC）。 */ + std::size_t edgeSerialChunkQueueCapacity{4096}; + /** false 时在同线程立即处理（调试用）；true 时走队列+线程（默认）。 */ + bool edgeUseEndpointFramerThreads{true}; + }; + + explicit EdgeStreamPipeline(Config cfg); + ~EdgeStreamPipeline(); + + bool ok() const { return m_ok; } + QString lastError() const { return m_lastError; } + + void setTreeModel(DeviceTreeModel* t) { m_treeModel = t; } + + /** 完整帧经此回调上行（通常绑定 `EdgeTcpUplinkPort::push`）。须在首包入队前设置。 */ + void setFramedSink(std::function sink); + + /** + * 将串口 read 组装的流式 chunk 入队（默认）或在同线程立即处理。 + * @return false 表示队列已满（与 daemon `chunkQ.push` 失败一致）。 + */ + bool tryEnqueueStreamChunk(softbus::core::models::RawBusMessage chunk); + + /** 分帧线程入口：framer → parser/mapper → framedSink（勿在串口回调线程直接调用）。 */ + void processFramerIngressFromWorker(softbus::core::models::RawBusMessage&& chunkIn); + +private: + void shutdownFramerWorkers(); + std::shared_ptr ensureFramerWorker(std::uint32_t endpointHash); + + void dispatchFramer(softbus::core::models::RawBusMessage& chunkIn, + std::vector& completeFramesOut); + std::vector runParserMapper(const softbus::core::models::RawBusMessage& ctx); + void attachEdgeContext(const softbus::core::models::RawBusMessage& raw, + std::vector& msgs) const; + + bool m_ok{false}; + QString m_lastError; + Config m_cfg; + DeviceTreeModel* m_treeModel{nullptr}; + std::shared_ptr m_pluginManager; + std::unique_ptr m_pluginHost; + std::unique_ptr m_passthroughFramer; + + std::mutex m_framerSessionMutex; + QHash> m_framerSessions; + + std::mutex m_parserSessionMutex; + QHash> m_parserSessions; + + std::mutex m_mapperSessionMutex; + QHash> m_mapperSessions; + + std::function m_framedSink; + mutable std::mutex m_framerWorkerMutex; + std::unordered_map> m_framerWorkers; +}; + +inline size_t qHash(const PluginSessionKey& key, size_t seed = 0) noexcept +{ + return ::qHash(key.endpointHash, ::qHash(key.pluginId, seed)); +} + +} // namespace softbus_edge diff --git a/src/utils/AgentDebugNdjson5448.h b/src/utils/AgentDebugNdjson5448.h new file mode 100644 index 0000000..1840479 --- /dev/null +++ b/src/utils/AgentDebugNdjson5448.h @@ -0,0 +1,35 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +// #region agent log +namespace agent_debug_5448 +{ +inline void log(const char* hypothesisId, const char* location, const char* message, const QJsonObject& data) +{ + QJsonObject root; + root.insert(QStringLiteral("sessionId"), QStringLiteral("5448fc")); + root.insert(QStringLiteral("hypothesisId"), QString::fromUtf8(hypothesisId)); + root.insert(QStringLiteral("location"), QString::fromUtf8(location)); + root.insert(QStringLiteral("message"), QString::fromUtf8(message)); + root.insert(QStringLiteral("data"), data); + root.insert(QStringLiteral("timestamp"), QDateTime::currentMSecsSinceEpoch()); + const QString path = + QStringLiteral("/home/dt/myCode/bus/soft_bus/softbus_daemon/.cursor/debug-5448fc.log"); + const QFileInfo fi(path); + QDir().mkpath(fi.absolutePath()); + QFile f(path); + if (!f.open(QIODevice::Append | QIODevice::Text)) { + return; + } + f.write(QJsonDocument(root).toJson(QJsonDocument::Compact)); + f.write("\n", 1); +} +} // namespace agent_debug_5448 +// #endregion