devices pipelinecontext
This commit is contained in:
flower_linux
@@ -1,14 +1,71 @@
|
||||
#include "message_bus/pipeline/PipelineEngine.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <iomanip>
|
||||
#include <sstream>
|
||||
#include <chrono>
|
||||
#include <functional>
|
||||
|
||||
#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h"
|
||||
#include "utils/logs/logging.h"
|
||||
|
||||
namespace softbus::message_bus::pipeline
|
||||
{
|
||||
|
||||
struct EndpointFramingState
|
||||
{
|
||||
softbus::core::memory::LockFreeQueue<PipelineContext> chunkQ;
|
||||
std::atomic<bool> running{true};
|
||||
std::thread th;
|
||||
std::uint64_t nextSeq{1};
|
||||
QString endpoint;
|
||||
std::shared_ptr<softbus::core::memory::MemoryPool> pool;
|
||||
softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig modbusCfg;
|
||||
std::shared_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> pass;
|
||||
std::function<void(PipelineContext)> pushFramed;
|
||||
|
||||
explicit EndpointFramingState(std::size_t cap)
|
||||
: chunkQ(cap)
|
||||
{
|
||||
}
|
||||
};
|
||||
|
||||
namespace
|
||||
{
|
||||
|
||||
void endpointFramerThread(std::shared_ptr<EndpointFramingState> st)
|
||||
{
|
||||
using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer;
|
||||
using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu;
|
||||
|
||||
auto modbus = std::make_unique<ModbusRtuFramer>(st->pool, st->modbusCfg, st->endpoint);
|
||||
|
||||
while (st->running.load(std::memory_order_acquire)) {
|
||||
PipelineContext ch;
|
||||
if (!st->chunkQ.pop(ch)) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
||||
continue;
|
||||
}
|
||||
std::vector<PipelineContext> out;
|
||||
if (protocolHintIsModbusRtu(ch.protocolHint)) {
|
||||
modbus->feed(ch, out);
|
||||
} else if (st->pass) {
|
||||
st->pass->feed(ch, out);
|
||||
}
|
||||
for (auto& x : out) {
|
||||
x.frameSeq = st->nextSeq++;
|
||||
if (st->pushFramed) {
|
||||
st->pushFramed(std::move(x));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
PipelineEngine::PipelineEngine(std::size_t queueCapacity)
|
||||
: m_ingressQ(queueCapacity), m_egressQ(queueCapacity)
|
||||
: m_ingressQ(queueCapacity)
|
||||
, m_egressQ(queueCapacity)
|
||||
{
|
||||
}
|
||||
|
||||
@@ -17,12 +74,73 @@ PipelineEngine::~PipelineEngine()
|
||||
stop();
|
||||
}
|
||||
|
||||
void PipelineEngine::setMemoryPool(std::shared_ptr<softbus::core::memory::MemoryPool> pool)
|
||||
{
|
||||
m_pool = std::move(pool);
|
||||
}
|
||||
|
||||
void PipelineEngine::setRuntimeConfig(PipelineRuntimeConfig cfg)
|
||||
{
|
||||
m_rtConfig = cfg;
|
||||
}
|
||||
|
||||
void PipelineEngine::loadRuntimeConfigFromJson(const QJsonObject& obj)
|
||||
{
|
||||
if (obj.isEmpty()) {
|
||||
LOG_ERROR() << "PipelineEngine: runtime config is empty";
|
||||
// return;
|
||||
}
|
||||
m_rtConfig.parallelPipeline = obj.value(QStringLiteral("parallelPipeline")).toBool(true);
|
||||
m_rtConfig.parseWorkerCount = std::max(1, obj.value(QStringLiteral("parseWorkerCount")).toInt(2));
|
||||
m_rtConfig.orderedEgress = obj.value(QStringLiteral("orderedEgress")).toBool(true);
|
||||
m_rtConfig.framedQueueMax = static_cast<std::size_t>(
|
||||
std::max(64, obj.value(QStringLiteral("framedQueueMax")).toInt(4096)));
|
||||
m_rtConfig.maxReorderDepth = static_cast<std::size_t>(
|
||||
std::max(16, obj.value(QStringLiteral("maxReorderDepth")).toInt(256)));
|
||||
m_rtConfig.modbus.maxAdu = static_cast<std::size_t>(
|
||||
std::max(8, obj.value(QStringLiteral("modbusRtuMaxFrameBytes")).toInt(256)));
|
||||
m_rtConfig.modbus.interFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0);
|
||||
}
|
||||
|
||||
void PipelineEngine::start()
|
||||
{
|
||||
if (m_running.exchange(true)) {
|
||||
return;
|
||||
}
|
||||
m_worker = std::thread([this]() { run(); });
|
||||
|
||||
if (!m_pool) {
|
||||
LOG_ERROR() << "PipelineEngine::start: memory pool not set";
|
||||
m_running = false;
|
||||
return;
|
||||
}
|
||||
|
||||
m_modbusFramer =
|
||||
std::make_unique<softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer>(m_pool,
|
||||
m_rtConfig.modbus,
|
||||
QString{});
|
||||
m_passthroughFramer =
|
||||
std::make_shared<softbus::message_bus::pipeline::stages::framers::PassthroughFramer>(m_pool);
|
||||
|
||||
// 如果启用并行管道 则创建帧队列和解析线程 如果启用有序出口 则创建合并队列和合并线程
|
||||
// 并行管道是用来处理多个端点的数据流的
|
||||
// 有序出口是用来保证数据流的有序性的
|
||||
// 帧队列是用来存储帧的
|
||||
// 解析线程是用来处理帧的
|
||||
// 合并队列是用来存储合并的帧的
|
||||
// 合并线程是用来处理合并的帧的
|
||||
if (m_rtConfig.parallelPipeline) {
|
||||
m_framedQ = std::make_unique<softbus::core::threading::MutexQueue<PipelineContext>>(
|
||||
m_rtConfig.framedQueueMax);
|
||||
for (int i = 0; i < m_rtConfig.parseWorkerCount; ++i) {
|
||||
m_parseThreads.emplace_back([this]() { parseWorkerLoop(); });
|
||||
}
|
||||
if (m_rtConfig.orderedEgress) {
|
||||
m_mergeQ = std::make_unique<softbus::core::threading::MutexQueue<MergeItem>>(0);
|
||||
m_mergerThread = std::thread([this]() { mergerLoop(); });
|
||||
}
|
||||
} else {
|
||||
m_worker = std::thread([this]() { runSingleWorker(); });
|
||||
}
|
||||
}
|
||||
|
||||
void PipelineEngine::stop()
|
||||
@@ -30,20 +148,77 @@ void PipelineEngine::stop()
|
||||
if (!m_running.exchange(false)) {
|
||||
return;
|
||||
}
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(m_endpointMutex);
|
||||
for (auto it = m_endpointFramers.constBegin(); it != m_endpointFramers.constEnd(); ++it) {
|
||||
if (it.value()) {
|
||||
it.value()->running.store(false, std::memory_order_release);
|
||||
}
|
||||
}
|
||||
for (auto it = m_endpointFramers.constBegin(); it != m_endpointFramers.constEnd(); ++it) {
|
||||
if (it.value() && it.value()->th.joinable()) {
|
||||
it.value()->th.join();
|
||||
}
|
||||
}
|
||||
m_endpointFramers.clear();
|
||||
}
|
||||
|
||||
if (m_worker.joinable()) {
|
||||
m_worker.join();
|
||||
}
|
||||
}
|
||||
|
||||
for (auto& t : m_parseThreads) {
|
||||
if (t.joinable()) {
|
||||
t.join();
|
||||
}
|
||||
}
|
||||
m_parseThreads.clear();
|
||||
|
||||
if (m_mergerThread.joinable()) {
|
||||
m_mergerThread.join();
|
||||
}
|
||||
|
||||
m_framedQ.reset();
|
||||
m_mergeQ.reset();
|
||||
m_modbusFramer.reset();
|
||||
m_passthroughFramer.reset();
|
||||
}
|
||||
// 功能 ：清空 ingress 队列
|
||||
// 参数 ：无
|
||||
// 返回值 ：无
|
||||
// 逻辑 ：从 ingress 队列中取出数据 直到队列为空
|
||||
void PipelineEngine::drain()
|
||||
{
|
||||
PipelineContext tmp;
|
||||
while (m_ingressQ.pop(tmp)) {
|
||||
}
|
||||
}
|
||||
|
||||
// 功能 ：将数据推送到 ingress 队列
|
||||
// 参数 ：ctx 数据
|
||||
// 返回值 ：是否成功
|
||||
// 逻辑 ：如果并行管道启用 则根据端点确保帧解析器 然后推送到帧队列 如果并行管道未启用 则直接推送到 ingress 队列
|
||||
bool PipelineEngine::enqueueIngress(PipelineContext ctx)
|
||||
{
|
||||
if (m_rtConfig.parallelPipeline) {
|
||||
if (ctx.endpoint.isEmpty()) {
|
||||
++m_counters.drop;
|
||||
return false;
|
||||
}
|
||||
ensureEndpointFramer(ctx.endpoint);
|
||||
std::shared_ptr<EndpointFramingState> st;
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(m_endpointMutex);
|
||||
st = m_endpointFramers.value(ctx.endpoint);
|
||||
}
|
||||
if (!st || !st->chunkQ.push(std::move(ctx))) {
|
||||
++m_counters.drop;
|
||||
return false;
|
||||
}
|
||||
++m_counters.in;
|
||||
return true;
|
||||
}
|
||||
|
||||
if (!m_ingressQ.push(std::move(ctx))) {
|
||||
++m_counters.drop;
|
||||
return false;
|
||||
@@ -51,41 +226,209 @@ bool PipelineEngine::enqueueIngress(PipelineContext ctx)
|
||||
++m_counters.in;
|
||||
return true;
|
||||
}
|
||||
|
||||
// 功能 ：将数据推送到 egress 队列
|
||||
// 参数 ：ctx 数据
|
||||
// 返回值 ：是否成功
|
||||
// 逻辑 ：将数据推送到 egress 队列
|
||||
bool PipelineEngine::enqueueEgress(PipelineContext ctx)
|
||||
{
|
||||
return m_egressQ.push(std::move(ctx));
|
||||
}
|
||||
|
||||
// 功能 ：尝试从 egress 队列中取出数据
|
||||
// 参数 ：out 数据
|
||||
// 返回值 ：是否成功
|
||||
// 逻辑 ：从 egress 队列中取出数据 如果队列为空 则返回false 如果队列不为空 则返回true
|
||||
bool PipelineEngine::tryPopEgress(PipelineContext& out)
|
||||
{
|
||||
return m_egressQ.pop(out);
|
||||
}
|
||||
|
||||
// 功能 ：设置插件管理器
|
||||
// 参数 ：pluginManager 插件管理器
|
||||
// 返回值 ：无
|
||||
// 逻辑 ：设置插件管理器
|
||||
void PipelineEngine::setPluginManager(std::shared_ptr<softbus::core::plugin_system::PluginManager> pluginManager)
|
||||
{
|
||||
m_pluginManager = std::move(pluginManager);
|
||||
}
|
||||
|
||||
// 管道引擎运行线程 从 ingress 队列中取出数据 进行处理 然后推送到 egress 队列中
|
||||
void PipelineEngine::run()
|
||||
// 功能 ：确保端点帧解析器
|
||||
// 参数 ：endpoint 端点
|
||||
// 返回值 ：无
|
||||
// 逻辑 ：如果端点帧解析器不存在 则创建端点帧解析器
|
||||
void PipelineEngine::ensureEndpointFramer(const QString& endpoint)
|
||||
{
|
||||
while (m_running.load()) {
|
||||
std::lock_guard<std::mutex> lk(m_endpointMutex);
|
||||
if (m_endpointFramers.contains(endpoint)) {
|
||||
return;
|
||||
}
|
||||
const std::size_t qcap = m_ingressQ.capacity();
|
||||
auto st = std::make_shared<EndpointFramingState>(qcap > 0 ? qcap : 4096);
|
||||
st->endpoint = endpoint;
|
||||
st->pool = m_pool;
|
||||
st->modbusCfg = m_rtConfig.modbus;
|
||||
st->pass = m_passthroughFramer;
|
||||
st->pushFramed = [this](PipelineContext c) {
|
||||
if (!m_framedQ) {
|
||||
return;
|
||||
}
|
||||
if (!m_framedQ->tryPush(std::move(c))) {
|
||||
++m_counters.drop;
|
||||
LOG_WARNING() << "PipelineEngine: framed queue full drop";
|
||||
}
|
||||
};
|
||||
st->th = std::thread(endpointFramerThread, st);
|
||||
m_endpointFramers.insert(endpoint, std::move(st));
|
||||
}
|
||||
// 功能 ：分发帧解析器
|
||||
// 参数 ：chunkIn 数据
|
||||
// 返回值 ：无
|
||||
// 逻辑 ：根据协议提示 选择合适的帧解析器 进行帧解析
|
||||
void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector<PipelineContext>& completeFramesOut)
|
||||
{
|
||||
using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu;
|
||||
// 根据协议提示 选择合适的帧解析器 进行帧解析
|
||||
|
||||
if (protocolHintIsModbusRtu(chunkIn.protocolHint)) {
|
||||
if (m_modbusFramer) {
|
||||
m_modbusFramer->feed(chunkIn, completeFramesOut);
|
||||
|
||||
}
|
||||
} else if (m_passthroughFramer) {
|
||||
m_passthroughFramer->feed(chunkIn, completeFramesOut);
|
||||
|
||||
}
|
||||
}
|
||||
// 功能 ：分配帧序列号
|
||||
void PipelineEngine::assignFrameSequence(PipelineContext& ctx)
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(m_seqMutex);
|
||||
ctx.frameSeq = ++m_nextFrameSeqByEndpoint[ctx.endpoint];
|
||||
}
|
||||
|
||||
void PipelineEngine::runSingleWorker()
|
||||
{
|
||||
// 单线程工作线程 从 ingress 队列中取出数据 进行处理 然后推送到 egress 队列中
|
||||
while (m_running.load(std::memory_order_acquire)) {
|
||||
PipelineContext ctx;
|
||||
if (!m_ingressQ.pop(ctx)) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
||||
continue;
|
||||
}
|
||||
|
||||
const bool framed = stage1Framer(ctx);
|
||||
// 帧解析
|
||||
const bool parsed = framed && stage2Parser(ctx);
|
||||
// 过滤
|
||||
const bool filtered = parsed && stage3Filter(ctx);
|
||||
// 验证
|
||||
const bool valid = filtered && stage4Validator(ctx);
|
||||
// 如果验证失败 则记录错误
|
||||
if (!valid) {
|
||||
std::vector<PipelineContext> batch;
|
||||
dispatchFramer(ctx, batch);
|
||||
|
||||
for (auto& c : batch) {
|
||||
if (c.frameKind != FrameKind::CompleteFrame) {
|
||||
continue;
|
||||
}
|
||||
//
|
||||
std::ostringstream oss;
|
||||
oss << std::hex << std::setfill('0');
|
||||
for (std::size_t i = 0; i < c.payload.block->size; ++i) {
|
||||
oss << std::setw(2) << static_cast<int>(c.payload.block->data[i]);
|
||||
if (i + 1 < c.payload.block->size) {
|
||||
oss << ' ';
|
||||
}
|
||||
}
|
||||
const QString hexPreview = QString::fromStdString(oss.str());
|
||||
LOG_INFO() << "the data is"<< hexPreview;
|
||||
// 分配帧序列号
|
||||
assignFrameSequence(c);
|
||||
if (!runStages234(c)) {
|
||||
++m_counters.error;
|
||||
continue;
|
||||
}
|
||||
if (!m_egressQ.push(std::move(c))) {
|
||||
++m_counters.drop;
|
||||
continue;
|
||||
}
|
||||
++m_counters.out;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void PipelineEngine::parseWorkerLoop()
|
||||
{
|
||||
while (true) {
|
||||
PipelineContext ctx;
|
||||
|
||||
const bool have = m_framedQ && m_framedQ->popWaitFor(ctx, m_running, std::chrono::milliseconds(50)); // 从帧队列中取出帧 , 如果帧队列为空 则等待50毫秒
|
||||
if (!have) {
|
||||
if (!m_running.load(std::memory_order_acquire)) {
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
std::ostringstream oss;
|
||||
oss << std::hex << std::setfill('0');
|
||||
for (std::size_t i = 0; i < ctx.payload.block->size; ++i) {
|
||||
oss << std::setw(2) << static_cast<int>(ctx.payload.block->data[i]);
|
||||
if (i + 1 < ctx.payload.block->size) {
|
||||
oss << ' ';
|
||||
}
|
||||
}
|
||||
const QString hexPreview = QString::fromStdString(oss.str());
|
||||
LOG_INFO() << "the data is"<< hexPreview;
|
||||
|
||||
const bool ok = runStages234(ctx);
|
||||
if (m_rtConfig.orderedEgress && m_mergeQ) {
|
||||
MergeItem it;
|
||||
it.endpoint = ctx.endpoint;
|
||||
it.seq = ctx.frameSeq;
|
||||
it.pipelineOk = ok;
|
||||
it.ctx = std::move(ctx);
|
||||
m_mergeQ->push(std::move(it));
|
||||
} else {
|
||||
if (!ok) {
|
||||
++m_counters.error;
|
||||
continue;
|
||||
}
|
||||
if (!m_egressQ.push(std::move(ctx))) {
|
||||
++m_counters.drop;
|
||||
continue;
|
||||
}
|
||||
++m_counters.out;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void PipelineEngine::mergerLoop()
|
||||
{
|
||||
while (true) {
|
||||
MergeItem it;
|
||||
const bool have = m_mergeQ && m_mergeQ->popWaitFor(it, m_running, std::chrono::milliseconds(50));
|
||||
if (!have) {
|
||||
if (!m_running.load(std::memory_order_acquire)) {
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
ingestMergeItem(std::move(it));
|
||||
}
|
||||
}
|
||||
// 功能 ：合并帧
|
||||
// 参数 ：item 合并项
|
||||
// 返回值 ：无
|
||||
// 逻辑 ：将合并项中的数据推送到 egress 队列中
|
||||
void PipelineEngine::ingestMergeItem(MergeItem item)
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(m_mergeStateMutex);
|
||||
auto& st = m_mergeByEndpoint[item.endpoint];
|
||||
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;
|
||||
return;
|
||||
}
|
||||
st.buffer.insert(item.seq, std::make_pair(std::move(item.ctx), item.pipelineOk));
|
||||
|
||||
while (st.buffer.contains(st.nextExpected)) {
|
||||
auto pr = st.buffer.take(st.nextExpected);
|
||||
++st.nextExpected;
|
||||
const bool pipelineOk = pr.second;
|
||||
PipelineContext ctx = std::move(pr.first);
|
||||
if (!pipelineOk) {
|
||||
++m_counters.error;
|
||||
continue;
|
||||
}
|
||||
@@ -97,17 +440,31 @@ void PipelineEngine::run()
|
||||
}
|
||||
}
|
||||
|
||||
bool PipelineEngine::stage1Framer(PipelineContext& ctx)
|
||||
// 功能 ：运行阶段234
|
||||
// 参数 ：ctx 数据
|
||||
// 返回值 ：是否成功
|
||||
// 逻辑 ：如果帧类型不是完整帧 则返回false 如果阶段2解析失败 则返回false 如果阶段3过滤失败 则返回false 如果阶段4验证失败 则返回false 如果都成功 则返回true
|
||||
bool PipelineEngine::runStages234(PipelineContext& ctx)
|
||||
{
|
||||
if (!ctx.payload.valid()) {
|
||||
if (ctx.frameKind != FrameKind::CompleteFrame) {
|
||||
return false;
|
||||
}
|
||||
if (ctx.payloadSize == 0) {
|
||||
ctx.payloadSize = ctx.payload.length;
|
||||
if (!stage2Parser(ctx)) {
|
||||
return false;
|
||||
}
|
||||
if (!stage3Filter(ctx)) {
|
||||
return false;
|
||||
}
|
||||
if (!stage4Validator(ctx)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// 功能 ：验证阶段4
|
||||
// 参数 ：ctx 数据
|
||||
// 返回值 ：是否成功
|
||||
// 逻辑 ：如果有效载荷无效 则返回false 如果都成功 则返回true
|
||||
bool PipelineEngine::stage4Validator(PipelineContext& ctx)
|
||||
{
|
||||
if (!ctx.payload.valid()) {
|
||||
@@ -119,9 +476,16 @@ bool PipelineEngine::stage4Validator(PipelineContext& ctx)
|
||||
if (!ctx.timestamp.isValid()) {
|
||||
ctx.timestamp = QDateTime::currentDateTimeUtc();
|
||||
}
|
||||
if (ctx.frameKind != FrameKind::CompleteFrame) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// 功能 ：运行阶段2解析器
|
||||
// 参数 ：ctx 数据
|
||||
// 返回值 ：是否成功
|
||||
// 逻辑 ：如果插件管理器为空 则返回true 如果插件管理器不为空 则选择合适的插件 创建会话 如果会话为空 则返回false 如果会话不为空 则返回true
|
||||
bool PipelineEngine::stage2Parser(PipelineContext& ctx)
|
||||
{
|
||||
if (!m_pluginManager) {
|
||||
|
||||
Reference in New Issue
Block a user