softbus_daemon/src/message_bus/pipeline/PipelineEngine.cpp

#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 "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.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)
{
}

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;
    }

    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()
{
    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;
    }
    ++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);
}

// 功能 ：确保端点帧解析器
// 参数 ：endpoint 端点
// 返回值 ：无
// 逻辑 ：如果端点帧解析器不存在 则创建端点帧解析器
void PipelineEngine::ensureEndpointFramer(const QString& endpoint)
{
    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;
        }

        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;
        }


        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;
        }
        if (!m_egressQ.push(std::move(ctx))) {
            ++m_counters.drop;
            continue;
        }
        ++m_counters.out;
    }
}

// 功能 ：运行阶段234
// 参数 ：ctx 数据
// 返回值 ：是否成功
// 逻辑 ：如果帧类型不是完整帧 则返回false 如果阶段2解析失败 则返回false 如果阶段3过滤失败 则返回false 如果阶段4验证失败 则返回false 如果都成功 则返回true
bool PipelineEngine::runStages234(PipelineContext& ctx)
{
    if (ctx.frameKind != FrameKind::CompleteFrame) {
        return false;
    }
    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;
    LOG_INFO() << "the protocol hint is"<< ctx.protocolHint;
    const QString directionText = (ctx.direction == BusDirection::Upstream)
                                      ? QStringLiteral("Upstream")
                                      : (ctx.direction == BusDirection::Downstream)
                                            ? QStringLiteral("Downstream")
                                            : QStringLiteral("Unknown");
    LOG_INFO() << "the direction is" << directionText;
    if (!stage2Parser(ctx)) {
        return false;
    }
    QJsonObject parsed = ctx.parsed;
    LOG_INFO() << "the parsed is"<< parsed;
    
    softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol = ctx.protocol;
    if (const auto* modbusPdu = std::get_if<softbus::message_bus::pipeline::protocol::ModbusRtuPdu>(&protocol.pdu)) {
        LOG_INFO() << "the protocol text is" << softbus::message_bus::pipeline::protocol::toText(*modbusPdu);
    } else {
        LOG_INFO() << "the protocol text is" << QStringLiteral("ProtocolEnvelopePdu::monostate");
    }
    // @TODO 以后需要测试是否丢包和解析延迟
    // m_count++;
    // LOG_INFO() << "the count of command is" << m_count;

    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()) {
        return false;
    }
    if (ctx.endpoint.isEmpty()) {
        return false;
    }
    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) {
        return true;
    }
    auto plugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint);
    if (!plugin) {
        return true;
    }
    auto session = plugin->createSession();
    if (!session) {
        return false;
    }
    return session->feed(ctx);
}

bool PipelineEngine::stage3Filter(PipelineContext& ctx)
{
    static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
    return s_filter.apply(ctx);
}

} // namespace softbus::message_bus::pipeline