diff --git a/bin/softbus_daemon b/bin/softbus_daemon index 6137844..f1609a4 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/docs/数据流转.md b/docs/数据流转.md index 5a1575c..798cc9a 100644 --- a/docs/数据流转.md +++ b/docs/数据流转.md @@ -4,7 +4,7 @@ - **物理设备**:真实硬件端口(当前主要是串口),由 `SerialQtDriver` 直接读写。 - **设备实例**:`SerialDevice`,负责单设备生命周期、驱动回调、上下行数据接入。 -- **原始总线消息**:`RawBusMessage`,包含 `endpoint`、`protocolHint`、`payload`、`traceId`、`direction` 等基础字段。 +- **原始总线消息**:`RawBusMessage`,包含 `endpointHash`、`protocol`、`routingKey`、`header`、`payload`、`traceId`、`direction` 等基础字段。 - **流水线上下文**:`PipelineContext`,继承 `RawBusMessage`,增加 `frameKind`、`frameSeq` 等管道字段。 - **业务消息**:经过协议解析和映射后得到 `DOMMessage`(统一语义数据)。 @@ -30,10 +30,12 @@ 1. 从 `MemoryPool` 分配内存块并拷贝原始字节。 2. 构造 `RawBusMessage`: - `direction = Upstream` - - `endpoint = 串口名` - - `protocolHint = DeviceInfo.protocol` - - `payload/payloadSize = 原始字节数据` - - `traceId = serial::` + - `endpointHash = makeEndpointHash(串口名)` + - `protocol = protocolTypeFromHint(DeviceInfo.protocol)` + - `routingKey = makeRoutingKey(endpointHash, deviceId)` + - `header = ProtocolHeader{}`(由后续 framer/parser 按协议填充 OOB 元数据) + - `payload = 原始字节数据`(长度由 `payload.length` 表示,不再维护 `payloadSize`) + - `traceId = serial::` 3. 调用 `m_ingress->push(std::move(ctx))` 进入统一入口。 4. 若内存池不足或 ingress 队列满,通过 `reportError(...)` 记录错误。 @@ -54,7 +56,7 @@ `enqueueIngress(...)` 分两种模式: - **并行模式(parallelPipeline=true)** - - 按 `endpoint` 创建/复用独立 `EndpointFramingState`(每端点一个分帧线程 + chunk 队列)。 + - 按 `endpointHash` 创建/复用独立 `EndpointFramingState`(每端点一个分帧线程 + chunk 队列)。 - chunk 先进入端点队列,端点线程负责分帧并赋值 `frameSeq`。 - 完整帧进入 `m_framedQ`,由解析线程池处理。 - **单线程模式(parallelPipeline=false)** @@ -64,9 +66,10 @@ `dispatchFramer(...)` 的选择策略: -1. 优先根据 `protocolHint` 从插件管理器选择协议分帧插件。 -2. 按 `endpoint + pluginId` 复用 framer session(保持协议状态机连续性)。 +1. 按 `protocol`(`ProtocolType`)从插件管理器选择协议分帧插件。 +2. 按结构化键 `SessionKey{endpointHash, pluginId}` 复用 framer/parser/mapper session(保持协议状态机连续性)。 3. 若没有可用插件,降级到 `PassthroughFramer`(透传/整包策略)。 +4. framer session 创建接口已升级为 `createFramerSession(uint32_t endpointHash, ...)`,不再传递 endpoint 字符串。 输出:`frameKind = CompleteFrame` 的帧批次。 @@ -84,6 +87,7 @@ - 最低校验当前为 `messageId` 非空。 4. **Transform/Enrich** - 仅对通过 filter+validator 的 `DOMMessage` 应用动态规则(scale/offset/unit)。 + - 规则作用域按 `endpointHash/deviceId/stableKey` 匹配,不再按 endpoint 字符串匹配。 - 写入 trace,补充富化字段。 说明(与代码一致): @@ -96,7 +100,7 @@ - 未开启有序出口:解析成功后直接进入 `m_egressQ`。 - 开启 `orderedEgress`: - 解析线程先写入 `m_mergeQ`。 - - `mergerLoop` 按 `endpoint + frameSeq` 重排。 + - `mergerLoop` 按 `endpointHash(映射键) + frameSeq` 重排。 - 使用 `nextExpected` 保证同端点有序输出。 - 超过 `maxReorderDepth` 的乱序缓存上限会丢弃并计数。 @@ -138,7 +142,7 @@ SerialQtDriver(read callback) ## 6. egress 到设备(现状) - `DeviceBusManager` 已启动 `processEgressLoop`,持续从 `PipelineEngine::tryPopEgress` 取数据并调用 `EgressRouter::send`。 -- `EgressRouter` 已实现 `endpoint -> SerialDevice` 查找和 `writePayload(...)` 发送。 +- `EgressRouter` 已实现基于 `endpointHash` 的 `SerialDevice` 查找和 `writePayload(...)` 发送。 - 当前关键缺口:`SerialDeviceManager::initializeSerialDeviceWithDeps` 中 `bindSerialDevice(...)` 仍被注释,导致路由表通常为空,`send` 多数返回失败。 - 因此当前链路是“egress 框架已存在,但默认没有绑定到串口实例”。 diff --git a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp index 51465e2..23ca7fb 100644 --- a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp +++ b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp @@ -17,10 +17,10 @@ class ModbusRtuFramerSession final : public softbus::core::plugin_system::IProto public: ModbusRtuFramerSession(std::shared_ptr pool, ModbusRtuFramerPluginConfig cfg, - QString endpoint) + std::uint32_t endpointHash) : m_pool(std::move(pool)) , m_cfg(cfg) - , m_endpoint(std::move(endpoint)) + , m_endpointHash(endpointHash) { m_buf.resize(m_cfg.maxAdu); } @@ -32,7 +32,7 @@ public: return; } const std::uint8_t* p = in.payload.bytes(); - const std::size_t n = in.payloadSize ? in.payloadSize : in.payload.length; + const std::size_t n = in.payload.length; if (!p || n == 0) { return; } @@ -47,7 +47,8 @@ public: m_lastActivity = m_lastRx; if (m_len + n > m_cfg.maxAdu) { - LOG_WARNING() << "ModbusRtuFramerPlugin: overflow endpoint=" << m_endpoint; + LOG_WARNING() << "ModbusRtuFramerPlugin: overflow endpointHash=0x" + << QString::number(m_endpointHash, 16); m_len = 0; return; } @@ -78,16 +79,22 @@ public: softbus::core::models::RawBusMessage framed; framed.deviceId = in.deviceId; - framed.stableKey = in.stableKey; - framed.endpoint = in.endpoint; - framed.timestamp = in.timestamp; - framed.protocolHint = in.protocolHint; + framed.endpointHash = in.endpointHash; + framed.protocol = softbus::core::models::ProtocolType::MODBUS_RTU; + framed.logicalAddress = in.logicalAddress; + framed.routingKey = in.routingKey; + framed.header = in.header; + framed.timestampMs = in.timestampMs; framed.traceId = in.traceId; framed.direction = in.direction; framed.payload.block = std::move(blk); framed.payload.offset = 0; framed.payload.length = *pred; - framed.payloadSize = *pred; + if (*pred >= 2 && framed.payload.block && framed.payload.block->data) { + framed.header.modbus.slaveId = framed.payload.block->data[0]; + framed.header.modbus.functionCode = framed.payload.block->data[1]; + framed.logicalAddress = framed.header.modbus.slaveId; + } out.push_back(std::move(framed)); const std::size_t rest = m_len - *pred; @@ -98,7 +105,6 @@ public: } in.payload = {}; - in.payloadSize = 0; } void reset() override { m_len = 0; } @@ -107,7 +113,7 @@ public: private: std::shared_ptr m_pool; ModbusRtuFramerPluginConfig m_cfg; - QString m_endpoint; + std::uint32_t m_endpointHash{0}; std::vector m_buf; std::size_t m_len{0}; std::chrono::steady_clock::time_point m_lastRx{std::chrono::steady_clock::now()}; @@ -128,9 +134,9 @@ QString ModbusRtuFramerPlugin::pluginId() const return QStringLiteral("modbus_rtu_framer"); } -bool ModbusRtuFramerPlugin::supports(const QString& protocolHint) const +softbus::core::models::ProtocolType ModbusRtuFramerPlugin::protocolType() const { - return isModbusRtuHint(protocolHint); + return softbus::core::models::ProtocolType::MODBUS_RTU; } void ModbusRtuFramerPlugin::bindMemoryPool(std::shared_ptr pool) @@ -139,7 +145,7 @@ void ModbusRtuFramerPlugin::bindMemoryPool(std::shared_ptr -ModbusRtuFramerPlugin::createFramerSession(const QString& endpoint, const QJsonObject& params) +ModbusRtuFramerPlugin::createFramerSession(std::uint32_t endpointHash, const QJsonObject& params) { if (!m_pool) { return {}; @@ -152,7 +158,7 @@ ModbusRtuFramerPlugin::createFramerSession(const QString& endpoint, const QJsonO if (params.contains(QStringLiteral("modbusRtuInterFrameTimeoutMs"))) { cfg.interFrameTimeoutMs = params.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(cfg.interFrameTimeoutMs); } - return std::make_unique(m_pool, cfg, endpoint); + return std::make_unique(m_pool, cfg, endpointHash); } std::shared_ptr makeModbusRtuFramerPlugin( diff --git a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h index 45b257b..0d330f6 100644 --- a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h +++ b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h @@ -27,10 +27,10 @@ public: ModbusRtuFramerPluginConfig cfg = {}); QString pluginId() const override; - bool supports(const QString& protocolHint) const override; + softbus::core::models::ProtocolType protocolType() const override; void bindMemoryPool(std::shared_ptr pool) override; std::unique_ptr createFramerSession( - const QString& endpoint, const QJsonObject& params) override; + std::uint32_t endpointHash, const QJsonObject& params) override; private: std::shared_ptr m_pool; diff --git a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp index eae4f86..d17a258 100644 --- a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp +++ b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp @@ -30,13 +30,14 @@ public: for (int i = 0; i < resp->registers.size(); ++i) { softbus::core::models::DOMMessage msg; msg.messageId = QStringLiteral("%1.reg.%2") - .arg(rawContext.stableKey.isEmpty() ? QString::number(rawContext.deviceId) - : rawContext.stableKey) + .arg(rawContext.routingKey == 0 + ? QString::number(rawContext.deviceId) + : QString::number(rawContext.routingKey)) .arg(i); - msg.timestamp = rawContext.timestamp.toMSecsSinceEpoch(); + msg.timestamp = rawContext.timestampMs; msg.value = static_cast(resp->registers[static_cast(i)]); msg.quality = softbus::core::models::DataQuality::GOOD; - msg.attributes.insert(QStringLiteral("sourceEndpoint"), rawContext.endpoint); + msg.attributes.insert(QStringLiteral("sourceEndpoint"), QString::number(rawContext.endpointHash)); out.push_back(std::move(msg)); } return out; @@ -46,12 +47,13 @@ public: std::get_if(pdu)) { softbus::core::models::DOMMessage msg; msg.messageId = QStringLiteral("%1.req.read_holding") - .arg(rawContext.stableKey.isEmpty() ? QString::number(rawContext.deviceId) - : rawContext.stableKey); - msg.timestamp = rawContext.timestamp.toMSecsSinceEpoch(); + .arg(rawContext.routingKey == 0 + ? QString::number(rawContext.deviceId) + : QString::number(rawContext.routingKey)); + msg.timestamp = rawContext.timestampMs; msg.value = static_cast(req->quantity); msg.quality = softbus::core::models::DataQuality::GOOD; - msg.attributes.insert(QStringLiteral("sourceEndpoint"), rawContext.endpoint); + msg.attributes.insert(QStringLiteral("sourceEndpoint"), QString::number(rawContext.endpointHash)); msg.attributes.insert(QStringLiteral("frameType"), QStringLiteral("request")); msg.attributes.insert(QStringLiteral("startAddress"), static_cast(req->startAddress)); msg.attributes.insert(QStringLiteral("quantity"), static_cast(req->quantity)); @@ -67,9 +69,9 @@ class ModbusRtuProtocolMapperPlugin final : public softbus::core::plugin_system: { public: QString pluginId() const override { return QStringLiteral("modbus_rtu_mapper"); } - bool supports(const QString& protocolHint) const override + softbus::core::models::ProtocolType protocolType() const override { - return isModbusRtuHint(protocolHint); + return softbus::core::models::ProtocolType::MODBUS_RTU; } std::unique_ptr createMapperSession() override { diff --git a/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp index f2dab6c..ce2393f 100644 --- a/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp +++ b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp @@ -55,7 +55,7 @@ public: if (!ctx.payload.valid()) { return false; } - const std::size_t n = ctx.payloadSize ? ctx.payloadSize : ctx.payload.length; + const std::size_t n = ctx.payload.length; const std::uint8_t* p = ctx.payload.bytes(); if (!p || n < 4) { return false; @@ -86,9 +86,9 @@ class ModbusRtuProtocolPlugin final : public IProtocolPlugin public: QString pluginId() const override { return QStringLiteral("modbus_rtu"); } - bool supports(const QString& protocolHint) const override + softbus::core::models::ProtocolType protocolType() const override { - return softbus::message_bus::pipeline::protocol::modbusrtu::isModbusRtuHint(protocolHint); + return softbus::core::models::ProtocolType::MODBUS_RTU; } std::unique_ptr createSession() override diff --git a/src/api/ipc_dbus/CommandDispatcher.cpp b/src/api/ipc_dbus/CommandDispatcher.cpp index d6b8da8..c8250ad 100644 --- a/src/api/ipc_dbus/CommandDispatcher.cpp +++ b/src/api/ipc_dbus/CommandDispatcher.cpp @@ -11,7 +11,7 @@ namespace softbus::message_bus::pipeline::RuleScope parseScope(const QJsonObject& params) { softbus::message_bus::pipeline::RuleScope scope; - scope.endpoint = params.value(QStringLiteral("endpoint")).toString(); + scope.endpointHash = static_cast(params.value(QStringLiteral("endpointHash")).toInteger(0)); scope.deviceId = params.value(QStringLiteral("deviceId")).toInt(-1); scope.stableKey = params.value(QStringLiteral("stableKey")).toString(); return scope; diff --git a/src/core/models/MessageRoutingUtils.h b/src/core/models/MessageRoutingUtils.h new file mode 100644 index 0000000..c48c5c1 --- /dev/null +++ b/src/core/models/MessageRoutingUtils.h @@ -0,0 +1,59 @@ +#pragma once + +#include + +#include + +#include "core/models/RawBusMessage.h" + +namespace softbus::core::models +{ + +// Hash physical endpoint string into a compact routing key. +inline std::uint32_t makeEndpointHash(const QString& endpoint) +{ + return qHash(endpoint); +} + +// Build a deterministic string key from endpoint hash (for session maps). +inline QString makeEndpointKey(std::uint32_t endpointHash) +{ + return QString::number(endpointHash, 16); +} + +// Build a 64-bit routing key for fast lookup paths. +inline std::uint64_t makeRoutingKey(std::uint32_t endpointHash, int deviceId) +{ + return (static_cast(endpointHash) << 32) | static_cast(deviceId); +} + +// Bridge routing key to legacy string-based matching interfaces. +inline QString routingKeyToStableKey(std::uint64_t routingKey) +{ + return routingKey == 0 ? QString() : QString::number(routingKey); +} + +inline ProtocolType protocolTypeFromHint(const QString& hint) +{ + if (hint.compare(QStringLiteral("modbus_rtu"), Qt::CaseInsensitive) == 0) { + return ProtocolType::MODBUS_RTU; + } + if (hint.compare(QStringLiteral("modbus_ascii"), Qt::CaseInsensitive) == 0) { + return ProtocolType::MODBUS_ASCII; + } + if (hint.compare(QStringLiteral("modbus_tcp"), Qt::CaseInsensitive) == 0) { + return ProtocolType::MODBUS_TCP; + } + if (hint.compare(QStringLiteral("can_raw"), Qt::CaseInsensitive) == 0) { + return ProtocolType::CAN_RAW; + } + if (hint.compare(QStringLiteral("can_open"), Qt::CaseInsensitive) == 0) { + return ProtocolType::CAN_OPEN; + } + if (hint.compare(QStringLiteral("udp_custom"), Qt::CaseInsensitive) == 0) { + return ProtocolType::UDP_CUSTOM; + } + return ProtocolType::UNKNOWN; +} + +} // namespace softbus::core::models diff --git a/src/core/models/RawBusMessage.h b/src/core/models/RawBusMessage.h index a3c9d04..1e7e5af 100644 --- a/src/core/models/RawBusMessage.h +++ b/src/core/models/RawBusMessage.h @@ -1,8 +1,7 @@ #pragma once -#include +#include -#include #include #include @@ -18,18 +17,61 @@ enum class BusDirection Downstream, }; +enum class ProtocolType : std::uint16_t +{ + UNKNOWN = 0, + MODBUS_RTU, + MODBUS_ASCII, + MODBUS_TCP, + CAN_RAW, + CAN_OPEN, + UDP_CUSTOM, +}; + +inline uint qHash(ProtocolType key, uint seed = 0) noexcept +{ + return ::qHash(static_cast(key), seed); +} + +union ProtocolHeader +{ + struct Can + { + std::uint32_t cobId{0}; + bool isExtended{false}; + bool isRtr{false}; + std::uint8_t padding[2]{0, 0}; + } can; + + struct Modbus + { + std::uint8_t slaveId{0}; + std::uint8_t functionCode{0}; + std::uint8_t padding[6]{0, 0, 0, 0, 0, 0}; + } modbus; + + struct Net + { + std::uint32_t srcIp{0}; + std::uint16_t srcPort{0}; + std::uint16_t padding{0}; + } net; + + std::uint64_t alignmentPadding{0}; +}; + struct RawBusMessage { - int deviceId{-1}; - QString endpoint; - QString stableKey; - QDateTime timestamp{QDateTime::currentDateTimeUtc()}; - QString protocolHint; + int deviceId{-1}; // Device instance id, -1 means unbound. + std::uint32_t endpointHash{0}; // Physical endpoint/channel hash. + ProtocolType protocol{ProtocolType::UNKNOWN}; + std::uint32_t logicalAddress{0}; // Slave ID / node ID for protocol routing. + std::uint64_t routingKey{0}; // Stable key for MetadataRegistry lookup. + ProtocolHeader header{}; // Out-of-band protocol header metadata. + std::int64_t timestampMs{0}; // UTC unix timestamp in milliseconds. QString traceId; softbus::core::models::PayloadRef payload; - std::size_t payloadSize{0}; - BusDirection direction{BusDirection::Unknown}; }; diff --git a/src/core/plugin_system/IProtocolFramerPlugin.h b/src/core/plugin_system/IProtocolFramerPlugin.h index a32f049..9116a4a 100644 --- a/src/core/plugin_system/IProtocolFramerPlugin.h +++ b/src/core/plugin_system/IProtocolFramerPlugin.h @@ -29,9 +29,9 @@ public: virtual ~IProtocolFramerPlugin() = default; virtual QString pluginId() const = 0; - virtual bool supports(const QString& protocolHint) const = 0; + virtual softbus::core::models::ProtocolType protocolType() const = 0; virtual void bindMemoryPool(std::shared_ptr pool) = 0; - virtual std::unique_ptr createFramerSession(const QString& endpoint, + virtual std::unique_ptr createFramerSession(std::uint32_t endpointHash, const QJsonObject& params) = 0; }; diff --git a/src/core/plugin_system/IProtocolMapperPlugin.h b/src/core/plugin_system/IProtocolMapperPlugin.h index 45aecdc..9902a01 100644 --- a/src/core/plugin_system/IProtocolMapperPlugin.h +++ b/src/core/plugin_system/IProtocolMapperPlugin.h @@ -28,7 +28,7 @@ public: virtual ~IProtocolMapperPlugin() = default; virtual QString pluginId() const = 0; - virtual bool supports(const QString& protocolHint) const = 0; + virtual softbus::core::models::ProtocolType protocolType() const = 0; virtual std::unique_ptr createMapperSession() = 0; }; diff --git a/src/core/plugin_system/IProtocolPlugin.h b/src/core/plugin_system/IProtocolPlugin.h index 020bd10..5c3ceb0 100644 --- a/src/core/plugin_system/IProtocolPlugin.h +++ b/src/core/plugin_system/IProtocolPlugin.h @@ -25,7 +25,7 @@ public: virtual ~IProtocolPlugin() = default; virtual QString pluginId() const = 0; - virtual bool supports(const QString& protocolHint) const = 0; + virtual softbus::core::models::ProtocolType protocolType() const = 0; virtual std::unique_ptr createSession() = 0; }; diff --git a/src/core/plugin_system/PluginManager.cpp b/src/core/plugin_system/PluginManager.cpp index 52afaee..2c415ff 100644 --- a/src/core/plugin_system/PluginManager.cpp +++ b/src/core/plugin_system/PluginManager.cpp @@ -8,22 +8,24 @@ void PluginManager::registerProtocolPlugin(std::shared_ptr plug if (!plugin) { return; } - m_protocolPlugins.insert(plugin->pluginId(), std::move(plugin)); + const auto protocol = plugin->protocolType(); + m_protocolPluginsByType.insert(protocol, plugin); + m_protocolPluginsById.insert(plugin->pluginId(), std::move(plugin)); } -std::shared_ptr PluginManager::selectProtocolPlugin(const QString& protocolHint) const +std::shared_ptr PluginManager::selectProtocolPlugin(softbus::core::models::ProtocolType protocol) const { - for (auto it = m_protocolPlugins.constBegin(); it != m_protocolPlugins.constEnd(); ++it) { - if (it.value() && it.value()->supports(protocolHint)) { - return it.value(); - } - } - return {}; + return m_protocolPluginsByType.value(protocol); } void PluginManager::unregisterProtocolPlugin(const QString& pluginId) { - m_protocolPlugins.remove(pluginId); + const auto it = m_protocolPluginsById.find(pluginId); + if (it == m_protocolPluginsById.end() || !it.value()) { + return; + } + m_protocolPluginsByType.remove(it.value()->protocolType()); + m_protocolPluginsById.erase(it); } void PluginManager::registerProtocolFramerPlugin(std::shared_ptr plugin) @@ -31,22 +33,25 @@ void PluginManager::registerProtocolFramerPlugin(std::shared_ptrpluginId(), std::move(plugin)); + const auto protocol = plugin->protocolType(); + m_protocolFramerPluginsByType.insert(protocol, plugin); + m_protocolFramerPluginsById.insert(plugin->pluginId(), std::move(plugin)); } -std::shared_ptr PluginManager::selectProtocolFramerPlugin(const QString& protocolHint) const +std::shared_ptr PluginManager::selectProtocolFramerPlugin( + softbus::core::models::ProtocolType protocol) const { - for (auto it = m_protocolFramerPlugins.constBegin(); it != m_protocolFramerPlugins.constEnd(); ++it) { - if (it.value() && it.value()->supports(protocolHint)) { - return it.value(); - } - } - return {}; + return m_protocolFramerPluginsByType.value(protocol); } void PluginManager::unregisterProtocolFramerPlugin(const QString& pluginId) { - m_protocolFramerPlugins.remove(pluginId); + const auto it = m_protocolFramerPluginsById.find(pluginId); + if (it == m_protocolFramerPluginsById.end() || !it.value()) { + return; + } + m_protocolFramerPluginsByType.remove(it.value()->protocolType()); + m_protocolFramerPluginsById.erase(it); } void PluginManager::registerProtocolMapperPlugin(std::shared_ptr plugin) @@ -54,37 +59,40 @@ void PluginManager::registerProtocolMapperPlugin(std::shared_ptrpluginId(), std::move(plugin)); + const auto protocol = plugin->protocolType(); + m_protocolMapperPluginsByType.insert(protocol, plugin); + m_protocolMapperPluginsById.insert(plugin->pluginId(), std::move(plugin)); } -std::shared_ptr PluginManager::selectProtocolMapperPlugin(const QString& protocolHint) const +std::shared_ptr PluginManager::selectProtocolMapperPlugin( + softbus::core::models::ProtocolType protocol) const { - for (auto it = m_protocolMapperPlugins.constBegin(); it != m_protocolMapperPlugins.constEnd(); ++it) { - if (it.value() && it.value()->supports(protocolHint)) { - return it.value(); - } - } - return {}; + return m_protocolMapperPluginsByType.value(protocol); } void PluginManager::unregisterProtocolMapperPlugin(const QString& pluginId) { - m_protocolMapperPlugins.remove(pluginId); + const auto it = m_protocolMapperPluginsById.find(pluginId); + if (it == m_protocolMapperPluginsById.end() || !it.value()) { + return; + } + m_protocolMapperPluginsByType.remove(it.value()->protocolType()); + m_protocolMapperPluginsById.erase(it); } QStringList PluginManager::registeredProtocolPluginIds() const { - return m_protocolPlugins.keys(); + return m_protocolPluginsById.keys(); } QStringList PluginManager::registeredProtocolFramerPluginIds() const { - return m_protocolFramerPlugins.keys(); + return m_protocolFramerPluginsById.keys(); } QStringList PluginManager::registeredProtocolMapperPluginIds() const { - return m_protocolMapperPlugins.keys(); + return m_protocolMapperPluginsById.keys(); } } // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/PluginManager.h b/src/core/plugin_system/PluginManager.h index 2f25c17..d98a4d5 100644 --- a/src/core/plugin_system/PluginManager.h +++ b/src/core/plugin_system/PluginManager.h @@ -19,22 +19,27 @@ public: PluginManager() = default; void registerProtocolPlugin(std::shared_ptr plugin); - std::shared_ptr selectProtocolPlugin(const QString& protocolHint) const; + std::shared_ptr selectProtocolPlugin(softbus::core::models::ProtocolType protocol) const; void unregisterProtocolPlugin(const QString& pluginId); void registerProtocolFramerPlugin(std::shared_ptr plugin); - std::shared_ptr selectProtocolFramerPlugin(const QString& protocolHint) const; + std::shared_ptr selectProtocolFramerPlugin( + softbus::core::models::ProtocolType protocol) const; void unregisterProtocolFramerPlugin(const QString& pluginId); void registerProtocolMapperPlugin(std::shared_ptr plugin); - std::shared_ptr selectProtocolMapperPlugin(const QString& protocolHint) const; + std::shared_ptr selectProtocolMapperPlugin( + softbus::core::models::ProtocolType protocol) const; void unregisterProtocolMapperPlugin(const QString& pluginId); QStringList registeredProtocolPluginIds() const; QStringList registeredProtocolFramerPluginIds() const; QStringList registeredProtocolMapperPluginIds() const; private: - QHash> m_protocolPlugins; - QHash> m_protocolFramerPlugins; - QHash> m_protocolMapperPlugins; + QHash> m_protocolPluginsById; + QHash> m_protocolPluginsByType; + QHash> m_protocolFramerPluginsById; + QHash> m_protocolFramerPluginsByType; + QHash> m_protocolMapperPluginsById; + QHash> m_protocolMapperPluginsByType; }; } // namespace softbus::core::plugin_system diff --git a/src/devices/physical/SerialDevice.cpp b/src/devices/physical/SerialDevice.cpp index 5ef6f64..f9aec83 100644 --- a/src/devices/physical/SerialDevice.cpp +++ b/src/devices/physical/SerialDevice.cpp @@ -7,6 +7,7 @@ #include #include "core/memory/MemoryPool.h" +#include "core/models/MessageRoutingUtils.h" #include "core/models/RawBusMessage.h" #include "hardware/drivers/serial/SerialQtDriver.h" #include "message_bus/ingress/IIngressPort.h" @@ -115,14 +116,17 @@ void SerialDevice::onRead(const std::vector& data) // 将数据推送到 ingress 队列 包含端口名、协议、数据、时间戳、traceId softbus::core::models::RawBusMessage ctx; ctx.direction = softbus::core::models::BusDirection::Upstream; - ctx.endpoint = m_config.endpoint; - ctx.protocolHint = m_info.protocol; + 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.payloadSize = data.size(); ctx.traceId = QStringLiteral("serial:%1:%2") .arg(m_config.endpoint) - .arg(QDateTime::currentMSecsSinceEpoch()); + .arg(ctx.timestampMs); if (!m_ingress->push(std::move(ctx))) { m_ingress->reportError(m_config.endpoint, QStringLiteral("ingress queue full")); diff --git a/src/message_bus/egress/EgressRouter.cpp b/src/message_bus/egress/EgressRouter.cpp index e0fae67..1a96efc 100644 --- a/src/message_bus/egress/EgressRouter.cpp +++ b/src/message_bus/egress/EgressRouter.cpp @@ -1,5 +1,6 @@ #include "message_bus/egress/EgressRouter.h" +#include "core/models/MessageRoutingUtils.h" #include "devices/physical/SerialDevice.h" namespace softbus::message_bus::egress @@ -19,8 +20,14 @@ void EgressRouter::unbindSerialDevice(const QString& endpoint) bool EgressRouter::send(const softbus::core::models::RawBusMessage& ctx) { - auto it = m_serialByEndpoint.constFind(ctx.endpoint); - if (it == m_serialByEndpoint.constEnd() || !it.value() || !ctx.payload.valid()) { + QSharedPointer device; + for (auto it = m_serialByEndpoint.constBegin(); it != m_serialByEndpoint.constEnd(); ++it) { + if (softbus::core::models::makeEndpointHash(it.key()) == ctx.endpointHash) { + device = it.value(); + break; + } + } + if (device.isNull() || !ctx.payload.valid()) { return false; } @@ -28,7 +35,7 @@ bool EgressRouter::send(const softbus::core::models::RawBusMessage& ctx) if (!p) { return false; } - return it.value()->writePayload(p, ctx.payload.length); + return device->writePayload(p, ctx.payload.length); } } // namespace softbus::message_bus::egress diff --git a/src/message_bus/pipeline/DynamicRuleRegistry.cpp b/src/message_bus/pipeline/DynamicRuleRegistry.cpp index 5fe72e7..f3f059b 100644 --- a/src/message_bus/pipeline/DynamicRuleRegistry.cpp +++ b/src/message_bus/pipeline/DynamicRuleRegistry.cpp @@ -45,13 +45,13 @@ bool DynamicRuleRegistry::removeTransformRule(const QString& id) } std::vector DynamicRuleRegistry::snapshotFilterRules( - const QString& endpoint, int deviceId, const QString& stableKey) const + std::uint32_t endpointHash, int deviceId, const QString& stableKey) const { std::shared_lock lk(m_mtx); std::vector out; out.reserve(static_cast(m_filterRules.size())); for (auto it = m_filterRules.constBegin(); it != m_filterRules.constEnd(); ++it) { - if (matchScope(it.value().scope, endpoint, deviceId, stableKey)) { + if (matchScope(it.value().scope, endpointHash, deviceId, stableKey)) { out.push_back(it.value()); } } @@ -59,13 +59,13 @@ std::vector DynamicRuleRegistry::snapshotFilterRules( } std::vector DynamicRuleRegistry::snapshotTransformRules( - const QString& endpoint, int deviceId, const QString& stableKey) const + std::uint32_t endpointHash, int deviceId, const QString& stableKey) const { std::shared_lock lk(m_mtx); std::vector out; out.reserve(static_cast(m_transformRules.size())); for (auto it = m_transformRules.constBegin(); it != m_transformRules.constEnd(); ++it) { - if (matchScope(it.value().scope, endpoint, deviceId, stableKey)) { + if (matchScope(it.value().scope, endpointHash, deviceId, stableKey)) { out.push_back(it.value()); } } @@ -82,13 +82,13 @@ QJsonObject DynamicRuleRegistry::status() const } bool DynamicRuleRegistry::matchScope( - const RuleScope& scope, const QString& endpoint, int deviceId, const QString& stableKey) const + const RuleScope& scope, std::uint32_t endpointHash, int deviceId, const QString& stableKey) const { - // hybrid scope: global + endpoint/device/stableKey selective match - const bool endpointOk = scope.endpoint.isEmpty() || scope.endpoint == endpoint; + // scope: global + endpointHash/device/stableKey selective match + const bool endpointHashOk = (scope.endpointHash == 0) || (scope.endpointHash == endpointHash); const bool deviceOk = scope.deviceId < 0 || scope.deviceId == deviceId; const bool stableKeyOk = scope.stableKey.isEmpty() || scope.stableKey == stableKey; - return endpointOk && deviceOk && stableKeyOk; + return endpointHashOk && deviceOk && stableKeyOk; } } // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/DynamicRuleRegistry.h b/src/message_bus/pipeline/DynamicRuleRegistry.h index 1bb6a2f..bda1298 100644 --- a/src/message_bus/pipeline/DynamicRuleRegistry.h +++ b/src/message_bus/pipeline/DynamicRuleRegistry.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include @@ -12,7 +13,7 @@ namespace softbus::message_bus::pipeline struct RuleScope { - QString endpoint; + std::uint32_t endpointHash{0}; int deviceId{-1}; QString stableKey; }; @@ -43,16 +44,14 @@ public: bool addTransformRule(const TransformRule& rule); bool removeTransformRule(const QString& id); - std::vector snapshotFilterRules( - const QString& endpoint, int deviceId, const QString& stableKey) const; - std::vector snapshotTransformRules( - const QString& endpoint, int deviceId, const QString& stableKey) const; + std::vector snapshotFilterRules(std::uint32_t endpointHash, int deviceId, const QString& stableKey) const; + std::vector snapshotTransformRules(std::uint32_t endpointHash, int deviceId, const QString& stableKey) const; QJsonObject status() const; private: DynamicRuleRegistry() = default; - bool matchScope(const RuleScope& scope, const QString& endpoint, int deviceId, const QString& stableKey) const; + bool matchScope(const RuleScope& scope, std::uint32_t endpointHash, int deviceId, const QString& stableKey) const; private: mutable std::shared_mutex m_mtx; diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index a85b23d..0bb3820 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -5,6 +5,7 @@ #include #include #include +#include "core/models/MessageRoutingUtils.h" #include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h" #include "utils/logs/logging.h" @@ -17,10 +18,10 @@ struct EndpointFramingState std::atomic running{true}; std::thread th; std::uint64_t nextSeq{1}; - QString endpoint; + std::uint32_t endpointHash{0}; std::shared_ptr pool; std::function&)> feed; - std::function pushFramed; + std::function pushFramed; explicit EndpointFramingState(std::size_t cap) : chunkQ(cap) @@ -46,7 +47,7 @@ void endpointFramerThread(std::shared_ptr st) for (auto& x : out) { const std::uint64_t seq = st->nextSeq++; if (st->pushFramed) { - st->pushFramed(st->endpoint, seq, std::move(x)); + st->pushFramed(st->endpointHash, seq, std::move(x)); } } } @@ -198,15 +199,15 @@ void PipelineEngine::drain() bool PipelineEngine::enqueueIngress(softbus::core::models::RawBusMessage ctx) { if (m_rtConfig.parallelPipeline) { - if (ctx.endpoint.isEmpty()) { + if (ctx.endpointHash == 0) { ++m_counters.drop; return false; } - ensureEndpointFramer(ctx.endpoint); + ensureEndpointFramer(ctx.endpointHash); std::shared_ptr st; { std::lock_guard lk(m_endpointMutex); - st = m_endpointFramers.value(ctx.endpoint); + st = m_endpointFramers.value(ctx.endpointHash); } if (!st || !st->chunkQ.push(std::move(ctx))) { ++m_counters.drop; @@ -252,25 +253,25 @@ void PipelineEngine::setPluginManager(std::shared_ptr lk(m_endpointMutex); - if (m_endpointFramers.contains(endpoint)) { + if (m_endpointFramers.contains(endpointHash)) { return; } const std::size_t qcap = m_ingressQ.capacity(); auto st = std::make_shared(qcap > 0 ? qcap : 4096); - st->endpoint = endpoint; + st->endpointHash = endpointHash; st->pool = m_pool; st->feed = [this](softbus::core::models::RawBusMessage& in, std::vector& out) { this->dispatchFramer(in, out); }; - st->pushFramed = [this](QString endpoint, std::uint64_t seq, softbus::core::models::RawBusMessage c) { + st->pushFramed = [this](std::uint32_t endpointHash, std::uint64_t seq, softbus::core::models::RawBusMessage c) { if (!m_framedQ) { return; } MergeItem item; - item.endpoint = std::move(endpoint); + item.endpointHash = endpointHash; item.seq = seq; item.ctx = std::move(c); item.pipelineOk = true; @@ -280,7 +281,7 @@ void PipelineEngine::ensureEndpointFramer(const QString& endpoint) } }; st->th = std::thread(endpointFramerThread, st); - m_endpointFramers.insert(endpoint, std::move(st)); + m_endpointFramers.insert(endpointHash, std::move(st)); } // 功能 :分发帧解析器 // 参数 :chunkIn 数据 @@ -290,14 +291,14 @@ void PipelineEngine::dispatchFramer( softbus::core::models::RawBusMessage& chunkIn, std::vector& completeFramesOut) { if (m_pluginManager) { - auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocolHint); + auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocol); if (plugin) { plugin->bindMemoryPool(m_pool); - const QString key = chunkIn.endpoint + QStringLiteral("|") + plugin->pluginId(); + const SessionKey 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.endpoint, QJsonObject{}); + auto s = plugin->createFramerSession(chunkIn.endpointHash, QJsonObject{}); m_framerSessions.insert( key, s ? std::shared_ptr(std::move(s)) : nullptr); it = m_framerSessions.find(key); @@ -313,10 +314,10 @@ void PipelineEngine::dispatchFramer( } } // 功能 :分配帧序列号 -std::uint64_t PipelineEngine::assignFrameSequence(const QString& endpoint) +std::uint64_t PipelineEngine::assignFrameSequence(std::uint32_t endpointHash) { std::lock_guard lk(m_seqMutex); - return ++m_nextFrameSeqByEndpoint[endpoint]; + return ++m_nextFrameSeqByEndpoint[endpointHash]; } void PipelineEngine::runSingleWorker() @@ -410,11 +411,12 @@ void PipelineEngine::mergerLoop() void PipelineEngine::ingestMergeItem(MergeItem item) { std::lock_guard lk(m_mergeStateMutex); - auto& st = m_mergeByEndpoint[item.endpoint]; + auto& st = m_mergeByEndpoint[item.endpointHash]; if (m_rtConfig.maxReorderDepth > 0 && static_cast(st.buffer.size()) >= m_rtConfig.maxReorderDepth) { ++m_counters.mergeDrop; - LOG_WARNING() << "PipelineEngine: merge reorder depth exceeded endpoint=" << item.endpoint; + LOG_WARNING() << "PipelineEngine: merge reorder depth exceeded endpointHash=0x" + << QString::number(item.endpointHash, 16); return; } st.buffer.insert(item.seq, std::make_pair(std::move(item.ctx), item.pipelineOk)); @@ -477,11 +479,11 @@ std::vector PipelineEngine::stage2Parser(cons if (!m_pluginManager) { return {}; } - auto parserPlugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint); + auto parserPlugin = m_pluginManager->selectProtocolPlugin(ctx.protocol); if (!parserPlugin) { return {}; } - const QString parserKey = ctx.endpoint + QStringLiteral("|") + parserPlugin->pluginId(); + const SessionKey parserKey{ctx.endpointHash, parserPlugin->pluginId()}; std::shared_ptr parserSession; { std::lock_guard lk(m_parserSessionMutex); @@ -503,21 +505,21 @@ std::vector PipelineEngine::stage2Parser(cons return {}; } - auto mapperPlugin = m_pluginManager->selectProtocolMapperPlugin(ctx.protocolHint); + auto mapperPlugin = m_pluginManager->selectProtocolMapperPlugin(ctx.protocol); if (!mapperPlugin) { return {}; } - const QString key = ctx.endpoint + QStringLiteral("|") + mapperPlugin->pluginId(); + const SessionKey mapperKey{ctx.endpointHash, mapperPlugin->pluginId()}; std::shared_ptr session; { std::lock_guard lk(m_mapperSessionMutex); - auto it = m_mapperSessions.find(key); + auto it = m_mapperSessions.find(mapperKey); if (it == m_mapperSessions.end() || !it.value()) { auto s = mapperPlugin->createMapperSession(); m_mapperSessions.insert( - key, + mapperKey, s ? std::shared_ptr(std::move(s)) : nullptr); - it = m_mapperSessions.find(key); + it = m_mapperSessions.find(mapperKey); } if (it == m_mapperSessions.end() || !it.value()) { return {}; @@ -537,7 +539,9 @@ softbus::core::models::EnrichedMessage PipelineEngine::applyTransformAndEnrich( const softbus::core::models::RawBusMessage& rawCtx, softbus::core::models::DOMMessage& message) const { auto rules = softbus::message_bus::pipeline::DynamicRuleRegistry::instance().snapshotTransformRules( - rawCtx.endpoint, rawCtx.deviceId, rawCtx.stableKey); + rawCtx.endpointHash, + rawCtx.deviceId, + softbus::core::models::routingKeyToStableKey(rawCtx.routingKey)); for (const auto& rule : rules) { const double base = std::visit([](const auto& x) { return static_cast(x); }, message.value); const double next = base * rule.scale + rule.offset; diff --git a/src/message_bus/pipeline/PipelineEngine.h b/src/message_bus/pipeline/PipelineEngine.h index ce8861a..3d8b983 100644 --- a/src/message_bus/pipeline/PipelineEngine.h +++ b/src/message_bus/pipeline/PipelineEngine.h @@ -25,6 +25,23 @@ namespace softbus::message_bus::pipeline { +struct SessionKey +{ + std::uint32_t endpointHash{0}; + QString pluginId; + + bool operator==(const SessionKey& other) const + { + return endpointHash == other.endpointHash && pluginId == other.pluginId; + } +}; + +inline uint qHash(const SessionKey& key, uint seed = 0) noexcept +{ + seed = ::qHash(key.endpointHash, seed); + return ::qHash(key.pluginId, seed); +} + struct PipelineRuntimeConfig { bool parallelPipeline{false}; @@ -70,7 +87,7 @@ public: private: struct MergeItem { - QString endpoint; + std::uint32_t endpointHash{0}; std::uint64_t seq{0}; softbus::core::models::RawBusMessage ctx; bool pipelineOk{false}; @@ -86,11 +103,11 @@ private: void parseWorkerLoop(); void mergerLoop(); - void ensureEndpointFramer(const QString& endpoint); + void ensureEndpointFramer(std::uint32_t endpointHash); void dispatchFramer(softbus::core::models::RawBusMessage& chunkIn, std::vector& completeFramesOut); - std::uint64_t assignFrameSequence(const QString& endpoint); + std::uint64_t assignFrameSequence(std::uint32_t endpointHash); bool runStages234(const softbus::core::models::RawBusMessage& ctx); void ingestMergeItem(MergeItem item); @@ -113,14 +130,14 @@ private: std::shared_ptr m_passthroughFramer; std::mutex m_framerSessionMutex; - QHash> m_framerSessions; + QHash> m_framerSessions; std::mutex m_parserSessionMutex; - QHash> m_parserSessions; + QHash> m_parserSessions; std::mutex m_mapperSessionMutex; - QHash> m_mapperSessions; + QHash> m_mapperSessions; std::mutex m_seqMutex; - QHash m_nextFrameSeqByEndpoint; + QHash m_nextFrameSeqByEndpoint; std::unique_ptr> m_framedQ; std::unique_ptr> m_mergeQ; @@ -128,10 +145,10 @@ private: std::thread m_mergerThread; std::mutex m_endpointMutex; - QHash> m_endpointFramers; + QHash> m_endpointFramers; std::mutex m_mergeStateMutex; - QHash m_mergeByEndpoint; + QHash m_mergeByEndpoint; int m_count{0}; // 计数器 }; diff --git a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp index f82a689..f44592f 100644 --- a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp +++ b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp @@ -16,7 +16,7 @@ void PassthroughFramer::feed(softbus::core::models::RawBusMessage& chu if (!m_pool || !chunkIn.payload.valid()) { return; } - const std::size_t n = chunkIn.payloadSize ? chunkIn.payloadSize : chunkIn.payload.length; + const std::size_t n = chunkIn.payload.length; if (n == 0) { return; } @@ -32,21 +32,21 @@ void PassthroughFramer::feed(softbus::core::models::RawBusMessage& chu blk->size = n; softbus::core::models::RawBusMessage out; - out.deviceId = chunkIn.deviceId; - out.stableKey = chunkIn.stableKey; - out.endpoint = chunkIn.endpoint; - out.timestamp = chunkIn.timestamp; - out.protocolHint = chunkIn.protocolHint; - out.traceId = chunkIn.traceId; - out.direction = chunkIn.direction; - out.payloadSize = n; + out.deviceId = chunkIn.deviceId; + out.endpointHash = chunkIn.endpointHash; + out.protocol = chunkIn.protocol; + out.logicalAddress = chunkIn.logicalAddress; + out.routingKey = chunkIn.routingKey; + out.header = chunkIn.header; + out.timestampMs = chunkIn.timestampMs; + out.traceId = chunkIn.traceId; + out.direction = chunkIn.direction; out.payload.block = std::move(blk); out.payload.offset = 0; out.payload.length = n; completeFramesOut.push_back(std::move(out)); chunkIn.payload = {}; - chunkIn.payloadSize = 0; } } // namespace softbus::message_bus::pipeline::stages::framers