VTrain/ai_matrix/framework/EngineManager.cpp

/**
 * engine管理实现
 * */

#include "EngineManager.h"
#include "Config.h"
#include "TimeUtil.h"
#include <curl/curl.h>

namespace ai_matrix
{
    EngineManager::EngineManager() {}

    EngineManager::~EngineManager() {}

    //初始化acl
    APP_ERROR EngineManager::Init()
    {
        if (!InitDeviceIds())
        {
            LogError << "InitDeviceIds err";
            return APP_ERR_COMM_INVALID_PARAM;
        }

        return APP_ERR_OK;
    }

    //去初始化acl
    APP_ERROR EngineManager::DeInit(void)
    {   
        return APP_ERR_OK;
    }

    //加载yaml文件中的配置
    APP_ERROR EngineManager::load_yaml_config(std::string path)
    {
        try
        {
            YAML::Node config = YAML::LoadFile(path);
            //退出程序
            if (config.IsNull())
            {
                LogError << "matrix.yaml err";
                return APP_ERR_COMM_INVALID_PARAM;
            }

            //engine使用deviceid
            mapUseDevice_["ALL"] = *init_deviceIds_.begin(); //默认所有engine使用初始化中最小deviceid
            if(config["use_deviceid"].IsDefined())
            {
                for (YAML::const_iterator it = config["use_deviceid"].begin(); it != config["use_deviceid"].end(); it++)
                {
                    std::string engineInfo = it->first.as<std::string>();
                    int deviceid = it->second.as<int>();
                    //使用deviceid必须是经过初始化的
                    if (init_deviceIds_.count(deviceid) == 0)
                    {
                        LogError << "use_deviceid set err value:" << deviceid;
                        return APP_ERR_COMM_INVALID_PARAM;
                    }
                    mapUseDevice_[engineInfo] = deviceid;
                }
            }

            //engine实例
            for (YAML::const_iterator it = config["engines"].begin(); it != config["engines"].end(); it++)
            {
                std::string engine_name = it->first.as<std::string>();
                int engine_id = it->second.as<int>();

                //检查是否有重复engine
                std::string engine_unique = engine_name + "_" + std::to_string(engine_id);
                auto iter = engine_map_.find(engine_unique);
                if (iter != engine_map_.end())
                {
                    continue;
                }

                //实例化engine
                std::shared_ptr<EngineBase> engineInstance = nullptr;
                EngineBase* base = (static_cast<EngineBase*>(EngineFactory::MakeEngine(engine_name)));
                if (base == nullptr)
                {
                    continue;
                }
                engineInstance.reset(base);

                //初始化engine
                APP_ERROR ret = InitEngineInstance(engineInstance, engine_name, engine_id);
                if (ret != APP_ERR_OK)
                {
                    continue;
                }

                //存入map
                engine_map_[engine_unique] = engineInstance;
            }

            //engine连接
            for (YAML::const_iterator it = config["connects"].begin(); it != config["connects"].end(); it++)
            {
                std::string from = it->first.as<std::string>();
                std::string to = it->second.as<std::string>();

                int iPos = to.find(" ");
                int iQueueSize = 0;
                if (iPos != std::string::npos)
                {
                    iQueueSize = atoi(to.substr(iPos+1, to.length()).c_str());
                    to = to.substr(0, iPos);
//                    LogInfo << "iQueueSize:" << iQueueSize;
                }

                LogInfo << "Add Connect,send:" << from << ",to:" << to;

                std::size_t pos = from.find_last_of("_");
                if (pos == std::string::npos)
                {
                    continue;
                }
                std::string src_engine = from.substr(0, pos);

                pos = to.find_last_of("_");
                if (pos == std::string::npos)
                {
                    continue;
                }
                std::string dst_engine = to.substr(0, pos);

                auto iterSend = engine_map_.find(src_engine);
                auto iterRecv = engine_map_.find(dst_engine);

                if (iterSend == engine_map_.end() || iterRecv == engine_map_.end())
                {
                    LogError << "Cann't find engine " << src_engine << " or " << dst_engine;
                    continue;
                }

                std::shared_ptr<MyQueue<std::shared_ptr<void>>> dataQueue = iterRecv->second->GetInputMap(to);
                if (dataQueue == nullptr)
                {
                    dataQueue = std::make_shared<MyQueue<std::shared_ptr<void>>>();
                    if (iQueueSize > 0)
                    {
                        dataQueue->setMaxSize(iQueueSize);
                    }

                    //设置engine输入队列
                    iterRecv->second->SetInputMap(to, dataQueue);

                    //设置engine输出队列
                    iterSend->second->SetOutputMap(from, dataQueue);
                }
                else
                {
                    //设置engine输出队列
                    iterSend->second->SetOutputMap(from, dataQueue);
                }
			}
        }
        catch (...) //捕获所有异常
        {
            return APP_ERR_COMM_INVALID_PARAM;
        }
        return APP_ERR_OK;
    }

    //初始化engine实例
    APP_ERROR EngineManager::InitEngineInstance(std::shared_ptr<EngineBase> engineInstance, std::string engineName, int engineId)
    {
        LogInfo << "EngineManager: begin to init engine instance,name=" << engineName << ", engine id = " << engineId << ".";

        //获取egnine使用的上下文
        std::string engineInfo = engineName + "_" + std::to_string(engineId);
        int deviceid;
        if(mapUseDevice_.count(engineInfo) > 0)
        {
            deviceid = mapUseDevice_[engineInfo];
        }
        else if(mapUseDevice_.count(std::to_string(engineId)) > 0)
        {
            deviceid = mapUseDevice_[std::to_string(engineId)];
        }
        else
        {
            deviceid = mapUseDevice_["ALL"];
        }

        EngineInitArguments initArgs;
        initArgs.deviceId = deviceId_;
        initArgs.engineName = engineName;
        initArgs.engineId = engineId;

        engineInstance->AssignInitArgs(initArgs); //填充参数

        APP_ERROR ret = engineInstance->Init(); //执行初始化
        if (ret != APP_ERR_OK)
        {
            LogError << "EngineManager: fail to init engine, name = " << engineName << ", engine id = " << engineId << ".";
            return ret;
        }
        LogInfo << "EngineManager: engine " << engineName << "[" << engineId << "] init success.";
        return ret;
    }

    //运行所有engine
    APP_ERROR EngineManager::RunAllEngine()
    {
        LogInfo << "begin to run engine.";
        for (auto it = engine_map_.begin(); it != engine_map_.end(); it++)
        {
            it->second->Run();
        }
        return APP_ERR_OK;
    }

    //停止所有engine
    APP_ERROR EngineManager::StopAllEngine()
    {
        LogInfo << "begin to stop engine.";
        for (auto it = engine_map_.begin(); it != engine_map_.end(); it++)
        {
            it->second->Stop();
        }
        return APP_ERR_OK;
    }

    //得到engine指针
    EngineBase *EngineManager::get_engine(std::string engineName)
    {
        auto iter = engine_map_.find(engineName);
        if (iter == engine_map_.end())
        {
            return nullptr;
        }
        return iter->second.get();
    }

    /**
    * 初始化设备id
    * inParam : N/A
    * outParam: N/A
    * return  : true(成功);false(失败)
    */
    bool EngineManager::InitDeviceIds()
    {
        init_deviceIds_.insert(0);
        return true;
    }
}