[rockit] 基于Google MediaPipe的PIPELINE多媒体框架设计总结_rockit框架

rockit 媒体框架借鉴Google MediaPipe的先进计算图理念, 完成了朴素版本的PIPELINE v1.0升级到基于TaskGraph的PIPELINE v2.0。

• 代码核心目录
  mediapipe源码：mediapipe
  rockit mediapipe代码: src/rt_task/task_graph
  google mediapipe源码目录：mediapipe/mediapipe/framework

• 代码流程图

APP-->new RTTaskGraph  // 创建图
-->RTTaskGraph::RTTaskGraph
    -->RTScheduler::RTScheduler  // 创建调度器Scheduler
        -->RTSchedulerQueue::RTSchedulerQueue  // 创建RTSchedulerQueue，并注册RTSchedulerShared结构体
        -->RTSchedulerQueue::setIdleCallback   // 注册回调RTScheduler::queueIdleStateChanged
        -->RTSchedulerQueue::setPauseCallback  // 注册回调RTScheduler::notifySchedulerPaused
    -->RTThreadPoolExecutor::RTThreadPoolExecutor // 同时创建RTExecutor，RTThreadPoolExecutor是RTExecutor的子类
        -->RTThreadPool::RTThreadPool  // 设置线程名称、数量、模式等信息
        -->RTThreadPoolExecutor::start  // 执行器开始工作
            -->RTThreadPool::startWorkers
                -->RTThreadPool::RTWorkerThread::RTWorkerThread
                    -->RtThread::RtThread // 创建RtThread对象
                    -->RtThread::start   // 创建线程
    -->RTScheduler::setDefaultExecutor  // 把执行器注册到调度器中  
1
2
3
4
5
6
7
8
9
10
11
12
13
14

APP-->initialize  // 加载json配置文件
-->RTTaskGraph::autoBuild
    -->RTNodeParser::RTNodeParser   // 创建RTNodeParser对象
    -->RTNodeParser::parseConfigFile  // 读取配置文件
        -->RTNodeParser::parseConfigText
    -->RTNodeParser::getNodeNum        // 解析配置文件的node数量
    -->RTNodeParser::getNodeIDByIndex  // 读取配置文件的node id
    -->RTTaskGraph::buildTaskNode      // 创建node
        -->RTNodeParser::initTaskNode    // 初始化当前node，
            -->node_proxy_parse_stub_uid   // 获取当前node的RTStubUid
            -->node_proxy_node_create      // 创建RTInputStreamManager和RTOutputStreamManager
                -->RTInputStreamManager::RTInputStreamManager
                -->RTOutputStreamManager::RTOutputStreamManager
                -->RT_NODE_FACTORY_CREATE_NODE  // 根据从配置中获取的RTStubUid创建RTTaskNode和插件node
                    -->RTTaskNodeFactory::createNode
                        -->RTTaskNode::RTTaskNode  // RTTaskNodeContextManager是的RTTaskNode成员变量（非指针），构造的时候会创建分配
                            -->RTTaskNodeContextManager::RTTaskNodeContextManager
                            -->RTInputStreamHandler::RTInputStreamHandler   // mInputHandler
                            -->RTOutputStreamHandler::RTOutputStreamHandler  // mOutputHandler
                            -->RtMetaData::RtMetaData   // mOptions
            -->RTNodeParser::syncOptions    // 同步配置文件的node的options：node_opts、stream_opts、node_opts_extra、stream_opts_extra
            -->RTInputStreamManager::initialize  // 同步配置文件的inputStream
            -->RTInputStreamManager::setStreamType
            -->RTOutputStreamManager::initialize  // 同步配置文件的outputStream
            -->RTOutputStreamManager::setStreamType  
            -->RTTaskNode::initialize      // 当前node初始化
                -->inputStreamInfos          // 创建RTStreamInfo用来保存inputStream信息
                -->RTStreamInfo::RTStreamInfo
                    -->RTStreamInfo::setStreamId
                    -->RTStreamInfo::setStreamName
                    -->RTStreamInfo::setStreamType
                -->RTInputStreamHandler::setInputStreamManager
                -->outputStreamInfos          // 创建RTStreamInfo用来保存ouputStream信息
                -->RTStreamInfo::RTStreamInfo
                    -->RTStreamInfo::setStreamId
                    -->RTStreamInfo::setStreamName
                    -->RTStreamInfo::setStreamType
                -->RTOutputStreamHandler::setOutputStreamManager
                -->RTTaskNodeContextManager::initialize
                    -->RTTaskNodeContext::RTTaskNodeContext
                        -->RTOutputStreamShared::RTOutputStreamShared   // 创建输出共享buffer
                            -->RTBufferListener::RTBufferListener      // buffer监听
                            -->RTMediaBufferPool::RTMediaBufferPool   // 创建buffer pool
    -->RTTaskGraph::autoLinkSource    // 连接source
        -->RTTaskNode::addStreamSource   //  
            -->RTOutputStreamManager::addMirror  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

APP-->prepare   // prepare
-->RTScheduler::flush
-->RTTaskGraph::prepareNodeForRun  // 每个node开始prepare
    -->TTaskNode::prepareRun  // 注册回调RTScheduler::addNodeToOpenedQueue和RTScheduler::scheduleNode
        -->RTInputStreamHandler ::prepareForRun  // 注册回调RTTaskNode::checkIfBecameReady()和RTScheduler::scheduleNode
        -->RTTaskNodeContextManager::prepareForRun
            -->RTTaskNodeContext::prepareForRun
                -->RTOutputStreamShared::prepareForRun
                    -->RTOutputStreamShared::allocAndRegisterBuffer    // 创建共享buffer
        -->RTTaskNode::open   // 调用各插件open
        -->RTTaskNode::nodeOpened  // 调用回调sourceNodeOpenedCallback, 即RTScheduler::addNodeToOpenedQueue
            -->RTScheduler::addNodeToOpenedQueue     // 
                -->RTTaskNode::setSchedulerQueue    //设置schedulerqueue
                -->RTScheduler::tryToScheduleIdleNode  // 当前图层所有的source添加到schedule队列中，以确保循环调度。暂停调度队列直到添加了所有的source.
                -->RTScheduler::setQueuesRunning(false)
                -->RTScheduler::scheduleUnthrottledNodes   //  添加所有的节点到调度器队列,
                    -->RTTaskNode::getDefaultNodeContext
                        -->RTTaskNodeContextManager::getDefaultNodeContext
                    -->RTTaskNodeContext::resume
                    -->RTTaskNode::checkIfBecameReady   // 检查SchedulingState
                        -->RTTaskNode::schedulingLoop   
                    // 调度循环线程，通用调度逻辑EndScheduling()(结束调度)和CheckIfBecameReady()(开启调度)， 
                   // 循环执行RTTaskNodeContext调度node，直到(1) 节点not ready (2)达到最大运行数
                            -->RTInputStreamHandler::scheduleInvocations  // 开启inputstream调度
                    -->RTScheduler::scheduleNode   // 
                        -->RTTaskNode::getSchedulerQueue
                            -->RTSchedulerQueue::addNode  //
                                -->RTSchedulerQueue::addItemToQueue
                                    -->RTSchedulerQueue::getTasksToSubmitToExecutor
                                    -->mIdleCallback  // RTScheduler::queueIdleStateChanged 回调
                                    -->RTTaskQueue::addTask
                -->RTScheduler::setQueuesRunning(true)
                -->RTScheduler::submitWaitingTasksOnQueues
                    -->RTSchedulerQueue::submitWaitingTasksToExecutor
                        -->RTSchedulerQueue::getTasksToSubmitToExecutor
                        -->RTTaskQueue::addTask
    -->RTTaskNode::isSource
    -->RTTaskNode::setQueueSizeCallbacks    // 绑定RTTaskGraph::updateThrottledNodes为伪函数，注册到RTInputStreamManager中
        -->RTInputStreamHandler::setQueueSizeCallbacks
            -->RTInputStreamManager::setQueueSizeCallbacks
    -->RTTaskNode::setMaxInputStreamQueueSize // TODO
    -->RTTaskNode::setMaxBatchPrcoessSize     // TODO             
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

APP-->start   // start
-->RTTaskGraph::startRun
    -->RTScheduler::start
        -->mState = STATE_RUNNING
        -->RTScheduler::setQueuesRunning(true)
            -->RTSchedulerQueue::setRunning(true)  // mRunningCount += running ? 1 : -1; mRunningCount为开关计数
                -->RTPausedCallback  // RTScheduler::notifySchedulerPaused的cb
        -->RTScheduler::handleIdle   // （Get the ball rolling）开始运作， while循环
            -->RTScheduler::tryToScheduleIdleNode  
1
2
3
4
5
6
7
8
9

APP-->stop
-->RTScheduler::stop   // stop
    -->sendInterrupt   // RTScheduler::notifyHasError
        -->RTScheduler::notifyHasError  // mShared.hasError = true
    -->mState = STATE_CANCELLING  //  
    -->RTScheduler::handleIdle   //  Because we have interrupt, this will cause the graph to quit.
    -->RTScheduler::submitWaitingTasksOnQueues
        -->RTSchedulerQueue::submitWaitingTasksToExecutor   // 如果调度器队列未执行时，有节点添加，不会立刻执行，先检查待等待执行的任务，并先执行先等待的任务。
1
2
3
4
5
6
7
8

mediapipe类关系图如下：