softbus_daemon/src/device_bus/manager/DeviceBusManager.cpp

#include "device_bus/manager/DeviceBusManager.h"

#include <chrono>
#include <memory>

#include <QSet>
// #include <algorithm>

#include "message_bus/egress/EgressRouter.h"
#include "message_bus/ingress/DriverIngressAdapter.h"
#include "message_bus/pipeline/PipelineEngine.h"
#include "core/plugin_system/PluginManager.h"
#include "device_bus/manager/SerialDeviceManager.h"
#include "devices/DeviceTypes.h"
#include "utils/logs/logging.h"

DeviceBusManager::~DeviceBusManager()
{
    shutdown();
}

void DeviceBusManager::setDeviceTreeModel(std::shared_ptr<DeviceTreeModel> model)
{
    m_treeModel = std::move(model);
}

void DeviceBusManager::setRegistry(softbus::device_bus::registry::IDeviceRegistry* registry)
{
    m_registry = registry;
}

void DeviceBusManager::setDirtyCallback(std::function<void()> cb)
{
    m_dirtyCb = std::move(cb);
}

bool DeviceBusManager::initialize(const QJsonObject& rootConfig)
{
    // 初始化内存池、pipeline引擎、插件管理器、ingress适配器、egress路由器
    // 内存池是用来存储数据块的
    // pipeline引擎是用来处理数据流的
    // 插件管理器是用来管理插件的
    // ingress适配器是用来将数据推送到 ingress 队列的
    // egress路由器是用来将数据从 ingress 队列中取出并发送给设备的
    // 内存池大小为4096，每个数据块大小为4096
    // pipeline引擎队列大小为4096
    // 插件管理器插件数量为4096
    // ingress适配器队列大小为4096
    // egress路由器队列大小为4096
    if (!m_memoryPool) {
        m_memoryPool = std::make_shared<softbus::core::memory::MemoryPool>(4096, 4096);
    }
    if (!m_pipelineEngine) {
        m_pipelineEngine = std::make_unique<softbus::message_bus::pipeline::PipelineEngine>(4096);
    }
    if (!m_pluginManager) {
        m_pluginManager = std::make_shared<softbus::core::plugin_system::PluginManager>();
    }
    if (m_pipelineEngine) {
        m_pipelineEngine->setPluginManager(m_pluginManager);
        m_pipelineEngine->start();
    }
    if (!m_ingressAdapter) {
        m_ingressAdapter = std::make_unique<softbus::message_bus::ingress::DriverIngressAdapter>(
            m_memoryPool, m_pipelineEngine.get());
    }
    if (!m_egressRouter) {
        m_egressRouter = std::make_unique<softbus::message_bus::egress::EgressRouter>();
    }
    if (!m_egressRunning.exchange(true)) {
        m_egressThread = std::thread([this]() { this->processEgressLoop(); });
    }

    if (!m_treeModel) {
        m_treeModel = std::make_shared<DeviceTreeModel>();
    }

    // 加载设备树
    QString treeError;
    if (!m_treeModel->loadFromDaemonConfig(rootConfig, &treeError)) {
        LOG_ERROR() << "DeviceBusManager: failed to load device_tree:" << treeError;
        return false;
    }

    const auto nodes = m_treeModel->nodes();
    for (const auto& n : nodes) {
        if (!n.enabled) continue;

        auto kind = softbus::devices::kindFromType(n.type);
        if (kind == softbus::devices::DeviceKind::Unknown) {
            kind = softbus::devices::inferDeviceKindFromEndpoint(n.endpoint);
        }

        if (kind == softbus::devices::DeviceKind::Serial) {
            initializeSerialDevice(n.endpoint, n.params);
            m_deviceEndpointsMap.insert(n.endpoint, kind);
        } else {
            LOG_WARNING() << "DeviceBusManager: skip unsupported kind for endpoint=" << n.endpoint
                          << "type=" << n.type;
        }
    }

    return true;
}

bool DeviceBusManager::reconcileFromTreeModel()
{
    if (!m_treeModel) {
        return false;
    }

    QSet<QString> desiredSerialEndpoints;
    const auto nodes = m_treeModel->nodes();
    for (const auto& n : nodes) {
        if (!n.enabled) continue;
        if (!n.params.value(QStringLiteral("online")).toBool(true)) continue;

        auto kind = softbus::devices::kindFromType(n.type);
        if (kind == softbus::devices::DeviceKind::Unknown) {
            kind = softbus::devices::inferDeviceKindFromEndpoint(n.endpoint);
        }

        if (kind == softbus::devices::DeviceKind::Serial) {
            desiredSerialEndpoints.insert(n.endpoint);
            if (!m_serialDevicesByEndpointSharedPtr.contains(n.endpoint)) {
                initializeSerialDevice(n.endpoint, n.params);
            }
            m_deviceEndpointsMap.insert(n.endpoint, kind);
        }
    }

    // Remove runtime devices that are no longer desired/offline
    const auto currentKeys = m_serialDevicesByEndpointSharedPtr.keys();
    for (const auto& endpoint : currentKeys) {
        if (!desiredSerialEndpoints.contains(endpoint)) {
            shutdownSerialDevice(endpoint);
            m_deviceEndpointsMap.remove(endpoint);
        }
    }

    return true;
}

void DeviceBusManager::shutdown()
{
    m_egressRunning.store(false);
    if (m_egressThread.joinable()) {
        m_egressThread.join();
    }

    // 关闭并销毁所有串口设备
    const auto keys = m_serialDevicesByEndpointSharedPtr.keys();
    for (const auto& endpoint : keys) {
        shutdownSerialDevice(endpoint);
    }
    if (m_pipelineEngine) {
        m_pipelineEngine->stop();
        m_pipelineEngine->drain();
    }
}

void DeviceBusManager::initializeSerialDevice(const QString& endpoint, const QJsonObject& params)
{
    softbus::device_bus::manager::SerialInitDeps deps;
    deps.treeModel = m_treeModel.get();
    deps.dirtyCb = m_dirtyCb;
    deps.registry = m_registry;
    deps.ingressAdapter = m_ingressAdapter.get();
    deps.memoryPool = m_memoryPool;
    deps.egressRouter = m_egressRouter.get();
    deps.serialDevicesByEndpoint = &m_serialDevicesByEndpointSharedPtr;
    (void)softbus::device_bus::manager::initializeSerialDeviceWithDeps(endpoint, params, deps);
}

void DeviceBusManager::shutdownSerialDevice(const QString& endpoint)
{
    auto it = m_serialDevicesByEndpointSharedPtr.find(endpoint);
    if (it == m_serialDevicesByEndpointSharedPtr.end()) {
        return;
    }

    auto device = it.value();
    if (device) {
        auto info = device->deviceInfo();
        if (m_registry && info.id >= 0) {
            m_registry->unregisterDevice(info.id);
        }
        device->stop();
    }
    if (m_egressRouter) {
        m_egressRouter->unbindSerialDevice(endpoint);
    }

    m_serialDevicesByEndpointSharedPtr.erase(it);
}

void DeviceBusManager::processEgressLoop()
{
    while (m_egressRunning.load()) {
        if (!m_pipelineEngine || !m_egressRouter) {
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
            continue;
        }

        softbus::message_bus::pipeline::PipelineContext out;
        if (!m_pipelineEngine->tryPopEgress(out)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            continue;
        }
        const bool ok = m_egressRouter->send(out);
        if (!ok) {
            // LOG_WARNING() << "DeviceBusManager: egress send failed endpoint=" << out.endpoint;
        }
    }
}