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

context bind

Browse Source
This commit is contained in:
flower_linux
2026-04-22 17:25:51 +08:00
parent 11bc015444
commit 7459cc0f36
31 changed files with 1427 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,7 +5,9 @@
#include <sstream>
#include <chrono>
#include <functional>
#include "core/models/MessageRoutingUtils.h"
#include "message_bus/pipeline/PipelineEngineApi.h"
#include "message_bus/pipeline/discovery/DiscoveryPool.h"
#include "message_bus/pipeline/mapping/MappingRegistry.h"
#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
@@ -287,6 +289,82 @@ void PipelineEngine::setPluginManager(std::shared_ptr<softbus::core::plugin_syst
m_pluginManager = std::move(pluginManager);
}
void PipelineEngine::resetCounters()
{
resetPipelineCounters(m_counters);
}
QJsonObject PipelineEngine::countersJson() const
{
QJsonObject out;
out.insert(QStringLiteral("in"), static_cast<qint64>(m_counters.in.load(std::memory_order_relaxed)));
out.insert(QStringLiteral("out"), static_cast<qint64>(m_counters.out.load(std::memory_order_relaxed)));
out.insert(QStringLiteral("drop"), static_cast<qint64>(m_counters.drop.load(std::memory_order_relaxed)));
out.insert(QStringLiteral("error"), static_cast<qint64>(m_counters.error.load(std::memory_order_relaxed)));
out.insert(QStringLiteral("mergeDrop"), static_cast<qint64>(m_counters.mergeDrop.load(std::memory_order_relaxed)));
return out;
}
QJsonObject PipelineEngine::runtimeConfigJson() const
{
QJsonObject out;
out.insert(QStringLiteral("parallelPipeline"), m_rtConfig.parallelPipeline);
out.insert(QStringLiteral("parseWorkerCount"), m_rtConfig.parseWorkerCount);
out.insert(QStringLiteral("orderedEgress"), m_rtConfig.orderedEgress);
out.insert(QStringLiteral("framedQueueMax"), static_cast<qint64>(m_rtConfig.framedQueueMax));
out.insert(QStringLiteral("maxReorderDepth"), static_cast<qint64>(m_rtConfig.maxReorderDepth));
return out;
}
QJsonObject PipelineEngine::status() const
{
int framerSessionCount = 0;
int parserSessionCount = 0;
int mapperSessionCount = 0;
int endpointFramerCount = 0;
int trackedEndpointCount = 0;
{
std::lock_guard<std::mutex> lk(m_framerSessionMutex);
framerSessionCount = m_framerSessions.size();
}
{
std::lock_guard<std::mutex> lk(m_parserSessionMutex);
parserSessionCount = m_parserSessions.size();
}
{
std::lock_guard<std::mutex> lk(m_mapperSessionMutex);
mapperSessionCount = m_mapperSessions.size();
}
{
std::lock_guard<std::mutex> lk(m_endpointMutex);
endpointFramerCount = m_endpointFramers.size();
}
{
std::lock_guard<std::mutex> lk(m_seqMutex);
trackedEndpointCount = m_nextFrameSeqByEndpoint.size();
}
QJsonObject out;
out.insert(QStringLiteral("running"), m_running.load(std::memory_order_acquire));
out.insert(QStringLiteral("runtimeConfig"), runtimeConfigJson());
out.insert(QStringLiteral("counters"), countersJson());
out.insert(QStringLiteral("parseWorkerCount"), static_cast<qint64>(m_parseThreads.size()));
out.insert(QStringLiteral("framerSessionCount"), framerSessionCount);
out.insert(QStringLiteral("parserSessionCount"), parserSessionCount);
out.insert(QStringLiteral("mapperSessionCount"), mapperSessionCount);
out.insert(QStringLiteral("endpointFramerCount"), endpointFramerCount);
out.insert(QStringLiteral("trackedEndpointCount"), trackedEndpointCount);
out.insert(QStringLiteral("ingressQueueCapacity"), static_cast<qint64>(m_ingressQ.capacity()));
out.insert(QStringLiteral("egressQueueCapacity"), static_cast<qint64>(m_egressQ.capacity()));
out.insert(QStringLiteral("hasPluginManager"), m_pluginManager != nullptr);
if (m_pluginManager) {
out.insert(QStringLiteral("parserPlugins"), QJsonArray::fromStringList(m_pluginManager->registeredProtocolPluginIds()));
out.insert(QStringLiteral("framerPlugins"), QJsonArray::fromStringList(m_pluginManager->registeredProtocolFramerPluginIds()));
out.insert(QStringLiteral("mapperPlugins"), QJsonArray::fromStringList(m_pluginManager->registeredProtocolMapperPluginIds()));
}
return out;
}
// 功能 ：确保端点帧解析器
// 参数 ：endpoint 端点
// 返回值 ：无
@@ -494,7 +572,18 @@ bool PipelineEngine::runStages234(const softbus::core::models::RawBusMessage& ct
if (!stage4Validator(msg)) {
continue;
}
(void)applyTransformAndEnrich(ctx, msg);
const auto enriched = applyTransformAndEnrich(ctx, msg);
const QString engineeringValueText = std::visit(
[](const auto& x) { return QString::number(static_cast<double>(x), 'g', 12); },
enriched.engineeringValue);
LOG_INFO() << "PipelineEngine: enriched message"
<< "sourceMessageId=" << enriched.sourceMessageId
<< "logicalName=" << enriched.logicalName
<< "engineeringValue=" << engineeringValueText
<< "unit=" << enriched.unit
<< "quality=" << static_cast<int>(enriched.quality)
<< "alarm=" << enriched.alarm
<< "traceCount=" << static_cast<int>(enriched.traceLog.size());
}
return true;
}
@@ -521,6 +610,7 @@ std::vector<softbus::core::models::DOMMessage> PipelineEngine::stage2Parser(cons
if (!parserPlugin) {
return {};
}
// 创建解析器会话
const SessionKey parserKey{ctx.endpointHash, parserPlugin->pluginId()};
std::shared_ptr<softbus::core::plugin_system::IProtocolSession> parserSession;
{

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

@@ -83,6 +83,10 @@ public:
void loadRuntimeConfigFromJson(const QJsonObject& obj);
const Counters& counters() const { return m_counters; }
void resetCounters();
QJsonObject countersJson() const;
QJsonObject runtimeConfigJson() const;
QJsonObject status() const;
private:
struct MergeItem
@@ -129,16 +133,16 @@ private:
PipelineRuntimeConfig m_rtConfig;
std::shared_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> m_passthroughFramer;
std::mutex m_framerSessionMutex;
mutable std::mutex m_framerSessionMutex;
QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>> m_framerSessions;
std::mutex m_parserSessionMutex;
mutable std::mutex m_parserSessionMutex;
QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolSession>> m_parserSessions;
std::mutex m_mapperSessionMutex;
mutable std::mutex m_mapperSessionMutex;
QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>> m_mapperSessions;
std::unique_ptr<softbus::core::plugin_system::IMappingLookup> m_mappingLookup;
std::unique_ptr<softbus::core::plugin_system::IDiscoverySink> m_discoverySink;
std::mutex m_seqMutex;
mutable std::mutex m_seqMutex;
QHash<std::uint32_t, std::uint64_t> m_nextFrameSeqByEndpoint;
std::unique_ptr<softbus::core::threading::MutexQueue<MergeItem>> m_framedQ;
@@ -146,7 +150,7 @@ private:
std::vector<std::thread> m_parseThreads;
std::thread m_mergerThread;
std::mutex m_endpointMutex;
mutable std::mutex m_endpointMutex;
QHash<std::uint32_t, std::shared_ptr<EndpointFramingState>> m_endpointFramers;
std::mutex m_mergeStateMutex;

15
src/message_bus/pipeline/PipelineEngineApi.cpp Normal file
View File

@@ -0,0 +1,15 @@
#include "message_bus/pipeline/PipelineEngineApi.h"
namespace softbus::message_bus::pipeline
{
void resetPipelineCounters(PipelineEngine::Counters& counters)
{
counters.in.store(0, std::memory_order_relaxed);
counters.out.store(0, std::memory_order_relaxed);
counters.drop.store(0, std::memory_order_relaxed);
counters.error.store(0, std::memory_order_relaxed);
counters.mergeDrop.store(0, std::memory_order_relaxed);
}
} // namespace softbus::message_bus::pipeline

10
src/message_bus/pipeline/PipelineEngineApi.h Normal file
View File

@@ -0,0 +1,10 @@
#pragma once
#include "message_bus/pipeline/PipelineEngine.h"
namespace softbus::message_bus::pipeline
{
void resetPipelineCounters(PipelineEngine::Counters& counters);
} // namespace softbus::message_bus::pipeline

50
src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp
View File

@@ -6,6 +6,32 @@
namespace softbus::message_bus::pipeline
{
namespace
{
// 规则里可能写完整 routingKey 的十进制串（与 routingKeyToStableKey 一致），也可能写 UI 侧常用的
// "ep_<十进制 endpointHash>"（仅按物理端点匹配，不含 deviceId 位）。
bool stableKeyMatches(const QString& pattern,
const QString& messageStableKey,
std::uint32_t endpointHash)
{
if (pattern.isEmpty()) {
return true;
}
if (pattern == messageStableKey) {
return true;
}
if (pattern.startsWith(QStringLiteral("ep_"), Qt::CaseInsensitive)) {
bool ok = false;
const QString digits = pattern.mid(3);
const quint32 parsed = digits.toUInt(&ok, 10);
return ok && parsed == endpointHash;
}
return false;
}
} // namespace
DynamicRuleRegistry& DynamicRuleRegistry::instance()
{
static DynamicRuleRegistry registry;
@@ -72,6 +98,28 @@ std::vector<TransformRule> DynamicRuleRegistry::snapshotTransformRules(
return out;
}
std::vector<FilterRule> DynamicRuleRegistry::snapshotAllFilterRules() 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) {
out.push_back(it.value());
}
return out;
}
std::vector<TransformRule> DynamicRuleRegistry::snapshotAllTransformRules() 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) {
out.push_back(it.value());
}
return out;
}
QJsonObject DynamicRuleRegistry::status() const
{
std::shared_lock<std::shared_mutex> lk(m_mtx);
@@ -86,7 +134,7 @@ bool DynamicRuleRegistry::matchScope(
{
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;
const bool stableKeyOk = stableKeyMatches(scope.stableKey, stableKey, endpointHash);
return endpointHashOk && deviceOk && stableKeyOk;
}

2
src/message_bus/pipeline/rules/DynamicRuleRegistry.h
View File

@@ -46,6 +46,8 @@ public:
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;
std::vector<FilterRule> snapshotAllFilterRules() const;
std::vector<TransformRule> snapshotAllTransformRules() const;
QJsonObject status() const;