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/3_filters/ProtocolParseFilter.h"
#include "utils/logs/logging.h"

namespace softbus::message_bus::pipeline
{

struct EndpointFramingState
{
    softbus::core::memory::LockFreeQueue<softbus::core::models::RawBusMessage> chunkQ;
    std::atomic<bool>                                   running{true};
    std::thread                                         th;
    std::uint64_t                                       nextSeq{1};
    QString                                             endpoint;
    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;

    explicit EndpointFramingState(std::size_t cap)
        : chunkQ(cap)
    {
    }
};

namespace
{

void endpointFramerThread(std::shared_ptr<EndpointFramingState> st)
{
    while (st->running.load(std::memory_order_acquire)) {
        softbus::core::models::RawBusMessage ch;
        if (!st->chunkQ.pop(ch)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            continue;
        }
        std::vector<softbus::core::models::RawBusMessage> out;
        if (st->feed) {
            st->feed(ch, out);
        }
        for (auto& x : out) {
            const std::uint64_t seq = st->nextSeq++;
            if (st->pushFramed) {
                st->pushFramed(st->endpoint, seq, 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)));
}

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_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<MergeItem>>(m_rtConfig.framedQueueMax);
        for (int i = 0; i < m_rtConfig.parseWorkerCount; ++i) {
            // 创建解析线程 解析线程从帧队列中取出帧 进行处理 
            // emplace_back 是用来创建解析线程的 参数是 lambda 表达式
            // [this]() { parseWorkerLoop(); } 是一个 lambda 表达式 用来创建解析线程
            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_passthroughFramer.reset();
    {
        std::lock_guard<std::mutex> lk(m_framerSessionMutex);
        m_framerSessions.clear();
    }
    {
        std::lock_guard<std::mutex> lk(m_parserSessionMutex);
        m_parserSessions.clear();
    }
    {
        std::lock_guard<std::mutex> lk(m_mapperSessionMutex);
        m_mapperSessions.clear();
    }
}
// 功能 ：清空 ingress 队列
// 参数 ：无
// 返回值 ：无
// 逻辑 ：从 ingress 队列中取出数据 直到队列为空
void PipelineEngine::drain()
{
    softbus::core::models::RawBusMessage tmp;
    while (m_ingressQ.pop(tmp)) {
    }
}
// 功能 ：将数据推送到 ingress 队列
// 参数 ：ctx 数据
// 返回值 ：是否成功
// 逻辑 ：如果并行管道启用 则根据端点确保帧解析器 然后推送到帧队列 如果并行管道未启用 则直接推送到 ingress 队列
bool PipelineEngine::enqueueIngress(softbus::core::models::RawBusMessage 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(softbus::core::models::RawBusMessage ctx)
{
    return m_egressQ.push(std::move(ctx));
}
// 功能 ：尝试从 egress 队列中取出数据
// 参数 ：out 数据
// 返回值 ：是否成功
// 逻辑 ：从 egress 队列中取出数据 如果队列为空 则返回false 如果队列不为空 则返回true
bool PipelineEngine::tryPopEgress(softbus::core::models::RawBusMessage& 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->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) {
        if (!m_framedQ) {
            return;
        }
        MergeItem item;
        item.endpoint = std::move(endpoint);
        item.seq = seq;
        item.ctx = std::move(c);
        item.pipelineOk = true;
        if (!m_framedQ->tryPush(std::move(item))) {
            ++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(
    softbus::core::models::RawBusMessage& chunkIn, std::vector<softbus::core::models::RawBusMessage>& completeFramesOut)
{
    if (m_pluginManager) {
        auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocolHint);
        if (plugin) {
            plugin->bindMemoryPool(m_pool);
            const QString key = chunkIn.endpoint + QStringLiteral("|") + 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{});
                m_framerSessions.insert(
                    key, s ? std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>(std::move(s)) : nullptr);
                it = m_framerSessions.find(key);
            }
            if (it != m_framerSessions.end() && it.value()) {
                it.value()->feedChunk(chunkIn, completeFramesOut);
                return;
            }
        }
    }
    if (m_passthroughFramer) {
        m_passthroughFramer->feed(chunkIn, completeFramesOut);
    }
}
// 功能 ：分配帧序列号
std::uint64_t PipelineEngine::assignFrameSequence(const QString& endpoint)
{
    std::lock_guard<std::mutex> lk(m_seqMutex);
    return ++m_nextFrameSeqByEndpoint[endpoint];
}

void PipelineEngine::runSingleWorker()
{
    // 单线程工作线程 从 ingress 队列中取出数据 进行处理 然后推送到 egress 队列中
    while (m_running.load(std::memory_order_acquire)) {
        softbus::core::models::RawBusMessage ctx;
        if (!m_ingressQ.pop(ctx)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            continue;
        }

        std::vector<softbus::core::models::RawBusMessage> batch;
        dispatchFramer(ctx, batch);

        for (auto& c : batch) {
            // 
            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;
            // 分配帧序列号
            const bool ok = runStages234(c);
            if (!ok) {
                ++m_counters.error;
                continue;
            }
            if (!m_egressQ.push(std::move(c))) {
                ++m_counters.drop;
                continue;
            }
            ++m_counters.out;
        }
    }
}

void PipelineEngine::parseWorkerLoop()
{
    while (true) {
        MergeItem item;
        const bool have = m_framedQ && m_framedQ->popWaitFor(item, m_running, std::chrono::milliseconds(50));
        if (!have) {
            if (!m_running.load(std::memory_order_acquire)) {
                break;
            }
            continue;
        }

        const bool ok = runStages234(item.ctx);
        if (m_rtConfig.orderedEgress && m_mergeQ) {
            item.pipelineOk = ok;
            m_mergeQ->push(std::move(item));
        } else {
            if (!ok) {
                ++m_counters.error;
                continue;
            }
            if (!m_egressQ.push(std::move(item.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;
        softbus::core::models::RawBusMessage 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(const softbus::core::models::RawBusMessage& ctx)
{
    auto messages = stage2Parser(ctx);
    LOG_INFO() << "PipelineEngine: stage2Parser messages size=" << messages.size();
    if (messages.empty()) {
        return false;
    }
    for (auto& msg : messages) {
        if (!stage3Filter(msg)) {
            continue;
        }
        if (!stage4Validator(msg)) {
            continue;
        }
        (void)applyTransformAndEnrich(ctx, msg);
    }
    return true;
}

// 功能 ：验证阶段4
// 参数 ：ctx 数据
// 返回值 ：是否成功
// 逻辑 ：如果有效载荷无效 则返回false 如果都成功 则返回true
bool PipelineEngine::stage4Validator(const softbus::core::models::DOMMessage& message) const
{
    return !message.messageId.isEmpty();
}

// 功能 ：运行阶段2解析器
// 参数 ：ctx 数据
// 返回值 ：是否成功
// 逻辑 ：如果插件管理器为空 则返回true 如果插件管理器不为空 则选择合适的插件 创建会话 如果会话为空 则返回false 如果会话不为空 则返回true
std::vector<softbus::core::models::DOMMessage> PipelineEngine::stage2Parser(const softbus::core::models::RawBusMessage& ctx)
{
    if (!m_pluginManager) {
        return {};
    }
    auto parserPlugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint);
    if (!parserPlugin) {
        return {};
    }
    const QString parserKey = ctx.endpoint + QStringLiteral("|") + parserPlugin->pluginId();
    std::shared_ptr<softbus::core::plugin_system::IProtocolSession> parserSession;
    {
        std::lock_guard<std::mutex> lk(m_parserSessionMutex);
        auto it = m_parserSessions.find(parserKey);
        if (it == m_parserSessions.end() || !it.value()) {
            auto s = parserPlugin->createSession();
            m_parserSessions.insert(
                parserKey,
                s ? std::shared_ptr<softbus::core::plugin_system::IProtocolSession>(std::move(s)) : nullptr);
            it = m_parserSessions.find(parserKey);
        }
        if (it == m_parserSessions.end() || !it.value()) {
            return {};
        }
        parserSession = it.value();
    }
    softbus::message_bus::pipeline::protocol::ProtocolEnvelope envelope; // 协议封装
    if (!parserSession->parse(ctx, envelope)) {
        return {};
    }

    auto mapperPlugin = m_pluginManager->selectProtocolMapperPlugin(ctx.protocolHint);
    if (!mapperPlugin) {
        return {};
    }
    const QString key = ctx.endpoint + QStringLiteral("|") + 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);
        if (it == m_mapperSessions.end() || !it.value()) {
            auto s = mapperPlugin->createMapperSession();
            m_mapperSessions.insert(
                key,
                s ? std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>(std::move(s)) : nullptr);
            it = m_mapperSessions.find(key);
        }
        if (it == m_mapperSessions.end() || !it.value()) {
            return {};
        }
        session = it.value();
    }
    return session->map(ctx, envelope);
}

bool PipelineEngine::stage3Filter(softbus::core::models::DOMMessage& message)
{
    static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
    return s_filter.process(message);
}

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);
    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;
        if (next != base) {
            message.appendTrace(QStringLiteral("transform:%1 %2->%3")
                                    .arg(rule.id)
                                    .arg(base, 0, 'g', 12)
                                    .arg(next, 0, 'g', 12));
        }
        message.value = next;
        if (!rule.unit.isEmpty()) {
            message.attributes.insert(QStringLiteral("unit"), rule.unit);
        }
    }
    static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
    return s_filter.enrich(message);
}

} // namespace softbus::message_bus::pipeline