flowzl/softbus_daemon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

endpointhash

Browse Source
This commit is contained in:
flower_linux
2026-04-15 20:05:04 +08:00
parent 94eec643de
commit 11bd86063d
21 changed files with 314 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/message_bus/pipeline/DynamicRuleRegistry.cpp
View File

@@ -45,13 +45,13 @@ bool DynamicRuleRegistry::removeTransformRule(const QString& id)
}
std::vector<FilterRule> DynamicRuleRegistry::snapshotFilterRules(
const QString& endpoint, int deviceId, const QString& stableKey) const
std::uint32_t endpointHash, int deviceId, const QString& stableKey) const
{
std::shared_lock<std::shared_mutex> lk(m_mtx);
std::vector<FilterRule> out;
out.reserve(static_cast<std::size_t>(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<FilterRule> DynamicRuleRegistry::snapshotFilterRules(
}
std::vector<TransformRule> DynamicRuleRegistry::snapshotTransformRules(
const QString& endpoint, int deviceId, const QString& stableKey) const
std::uint32_t endpointHash, int deviceId, const QString& stableKey) const
{
std::shared_lock<std::shared_mutex> lk(m_mtx);
std::vector<TransformRule> out;
out.reserve(static_cast<std::size_t>(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

11
src/message_bus/pipeline/DynamicRuleRegistry.h
View File

@@ -1,5 +1,6 @@
#pragma once
#include <cstdint>
#include <shared_mutex>
#include <vector>
@@ -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<FilterRule> snapshotFilterRules(
const QString& endpoint, int deviceId, const QString& stableKey) const;
std::vector<TransformRule> snapshotTransformRules(
const QString& endpoint, int deviceId, const QString& stableKey) const;
std::vector<FilterRule> snapshotFilterRules(std::uint32_t endpointHash, int deviceId, const QString& stableKey) const;
std::vector<TransformRule> 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;

58
src/message_bus/pipeline/PipelineEngine.cpp
View File

@@ -5,6 +5,7 @@
#include <sstream>
#include <chrono>
#include <functional>
#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<bool> running{true};
std::thread th;
std::uint64_t nextSeq{1};
QString endpoint;
std::uint32_t endpointHash{0};
std::shared_ptr<softbus::core::memory::MemoryPool> pool;
std::function<void(softbus::core::models::RawBusMessage&, std::vector<softbus::core::models::RawBusMessage>&)> feed;
std::function<void(QString, std::uint64_t, softbus::core::models::RawBusMessage)> pushFramed;
std::function<void(std::uint32_t, std::uint64_t, softbus::core::models::RawBusMessage)> pushFramed;
explicit EndpointFramingState(std::size_t cap)
: chunkQ(cap)
@@ -46,7 +47,7 @@ void endpointFramerThread(std::shared_ptr<EndpointFramingState> 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<EndpointFramingState> st;
{
std::lock_guard<std::mutex> 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<softbus::core::plugin_syst
// 参数 ：endpoint 端点
// 返回值 ：无
// 逻辑 ：如果端点帧解析器不存在 则创建端点帧解析器
void PipelineEngine::ensureEndpointFramer(const QString& endpoint)
void PipelineEngine::ensureEndpointFramer(std::uint32_t endpointHash)
{
std::lock_guard<std::mutex> 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<EndpointFramingState>(qcap > 0 ? qcap : 4096);
st->endpoint = endpoint;
st->endpointHash = endpointHash;
st->pool = m_pool;
st->feed = [this](softbus::core::models::RawBusMessage& in, std::vector<softbus::core::models::RawBusMessage>& 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<softbus::core::models::RawBusMessage>& 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<std::mutex> 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<softbus::core::plugin_system::IProtocolFramerSession>(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<std::mutex> 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<std::mutex> lk(m_mergeStateMutex);
auto& st = m_mergeByEndpoint[item.endpoint];
auto& st = m_mergeByEndpoint[item.endpointHash];
if (m_rtConfig.maxReorderDepth > 0
&& static_cast<std::size_t>(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<softbus::core::models::DOMMessage> 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<softbus::core::plugin_system::IProtocolSession> parserSession;
{
std::lock_guard<std::mutex> lk(m_parserSessionMutex);
@@ -503,21 +505,21 @@ std::vector<softbus::core::models::DOMMessage> 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<softbus::core::plugin_system::IProtocolMapperSession> session;
{
std::lock_guard<std::mutex> 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<softbus::core::plugin_system::IProtocolMapperSession>(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<double>(x); }, message.value);
const double next = base * rule.scale + rule.offset;

35
src/message_bus/pipeline/PipelineEngine.h
View File

@@ -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<softbus::core::models::RawBusMessage>& 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<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> m_passthroughFramer;
std::mutex m_framerSessionMutex;
QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>> m_framerSessions;
QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>> m_framerSessions;
std::mutex m_parserSessionMutex;
QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolSession>> m_parserSessions;
QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolSession>> m_parserSessions;
std::mutex m_mapperSessionMutex;
QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>> m_mapperSessions;
QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>> m_mapperSessions;
std::mutex m_seqMutex;
QHash<QString, std::uint64_t> m_nextFrameSeqByEndpoint;
QHash<std::uint32_t, std::uint64_t> m_nextFrameSeqByEndpoint;
std::unique_ptr<softbus::core::threading::MutexQueue<MergeItem>> m_framedQ;
std::unique_ptr<softbus::core::threading::MutexQueue<MergeItem>> m_mergeQ;
@@ -128,10 +145,10 @@ private:
std::thread m_mergerThread;
std::mutex m_endpointMutex;
QHash<QString, std::shared_ptr<EndpointFramingState>> m_endpointFramers;
QHash<std::uint32_t, std::shared_ptr<EndpointFramingState>> m_endpointFramers;
std::mutex m_mergeStateMutex;
QHash<QString, MergeState> m_mergeByEndpoint;
QHash<std::uint32_t, MergeState> m_mergeByEndpoint;
int m_count{0}; // 计数器
};

20
src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp
View File

@@ -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