x265探索与研究（八）：x265中的并行处理机制函数关系分析

        HEVC的高计算复杂度如果仅仅依赖于单核处理器计算能力的提高，其代价是非常昂贵的，为此，HEVC的设计充分考虑到了并行计算的需求。x265不同于HM（HEVC test Model），x265将并行计算发挥地更加淋淋尽致。在main()函数中，为了完成多线程计算，读完24帧输入帧后才开始编码的原因也基于此。

        为了理清x265中的并行处理机制，首先给出了如下图的并行处理函数关系图：

        经过前面几篇文章的分析，我们知道main()函数主要调用了encoder_open()函数、encoder_headers()函数、encoder_encode()函数与encoder_close()函数。其中：

（1）encoder_encode()函数调用了encode()函数，而encode()函数中调用了startCompressFrame()函数，在startCompressFrame()函数中，采用m_enable.trigger完成了触发线程的功能；

（2）encoder_open()函数调用了encoder_create()函数，在encoder_create()函数中等待线程的初始化并进入threadMain()函数中等待线程的触发，线程一旦触发则调用compressFrame函数进行进一步地编码工作。

        这两步的协调工作完美地成就了多线程的实现。为了进一步分析HEVC的并行处理机制，接下来首先分析视频编码的并行处理相关技术，然后依次分析该函数关系图中较为重要的startCompressFrame()函数、encoder_create()函数以及threadMain()函数。

1、并行处理技术

        并行处理一般指许多指令得以同时进行的处理模式，通常分为两种：功能并行和数据并行。

（1）功能并行是指将应用程序划分成互相独立的功能模块，每个模块间可以并行地执行，这种并行方式也称为流水线型并行，它将各个独立的模块划分给不同的运算单元，各个模块之间通过流的方式来进行数据交换和通信，最终再将各个单元串接在一起。功能并行充分利用了时间上的并行性来获得加速的效果，比较适合于硬件实现。功能并行的缺点是很明显的，a、由于分配给不同运算单元的功能模块是不同的，因此很容易产生载荷失衡问题；b、功能并行还需要在不同运算单元间进行数据通信，当数据量较大时，需要花费额外的资源来进行存储；c、另外，功能并行的扩展性较差。

（2）数据并行是将数据信息划分为相互独立的部分，每一部分交给不同的运算单元来执行，从而实现并行处理。在这种方式下，不同运算单元上执行的程序是相同的，而且处理的是相互独立的数据信息，因此不需要进行运算单元间的通信；数据并行也具有较好的扩展性，易于软件实现。

        HEVC/H.265提供了适用于进行数据并行处理的结构单元，如片和Tile，在不同的片和Tile中，数据信息是相互独立的，这样有利于将其分配给不同的运算单元来处理；对于小于片和Tile的划分单元，HEVC支持波前并行处理（Wavefront Parallel Processing, WPP），这是对于相互具有依赖关系的图像单元进行数据并行处理的方法。在HEVC中，并行处理技术主要包括：基于Tile的并行和波前并行两种。在进行基于Tile的并行时，由于Tile的相互独立性，不需要考虑它们之间的相互依赖关系，而在进行波前并行处理时，数据间的相互依赖关系是必不可少的。

2、startCompressFrame()函数的分析

        startCompressFrame()函数的主要功能就是触发线程，对应的代码分析如下：

/*=============================================================*/
/*
 ====== Analysed by: RuiDong Fang 
 ====== Csdn Blog:	 http://blog.csdn.net/frd2009041510 
 ====== Date:		 2016.04.15
 ====== Funtion：	 startCompressFrame()函数,触发线程。
 */
/*=============================================================*/
bool FrameEncoder::startCompressFrame(Frame* curFrame)
{
    m_slicetypeWaitTime = x265_mdate() - m_prevOutputTime;
    m_frame = curFrame;
    m_param = curFrame->m_param;
    m_sliceType = curFrame->m_lowres.sliceType;
    curFrame->m_encData->m_frameEncoderID = m_jpId;
    curFrame->m_encData->m_jobProvider = this;
    curFrame->m_encData->m_slice->m_mref = m_mref;

    if (!m_cuGeoms)
    {
        if (!initializeGeoms())
            return false;
    }

    m_enable.trigger();	//触发线程，下一步将会进入threadMain()函数中
    return true;
}

3、encoder_create()函数的分析

        encoder_create()函数的主要功能是检测线程池、可用的线程数目等等，若线程使用的条件符合则调用threadMain()函数，对应的代码分析如下：

void Encoder::create()
{
    if (!primitives.pu[0].sad)
    {
        // this should be an impossible condition when using our public API, and indicates a serious bug.
        x265_log(m_param, X265_LOG_ERROR, "Primitives must be initialized before encoder is created\n");
        abort();
    }

    x265_param* p = m_param;

    int rows = (p->sourceHeight + p->maxCUSize - 1) >> g_log2Size[p->maxCUSize];
    int cols = (p->sourceWidth  + p->maxCUSize - 1) >> g_log2Size[p->maxCUSize];

    // Do not allow WPP if only one row or fewer than 3 columns, it is pointless and unstable
	//对于不符合条件的，不进行WPP
    if (rows == 1 || cols < 3)
    {
        x265_log(p, X265_LOG_WARNING, "Too few rows/columns, --wpp disabled\n");
        p->bEnableWavefront = 0;
    }

    bool allowPools = !p->numaPools || strcmp(p->numaPools, "none");

    // Trim the thread pool if --wpp, --pme, and --pmode are disabled
	//如果--wpp, --pme, and --pmode不使能,清理线程池
    if (!p->bEnableWavefront && !p->bDistributeModeAnalysis && !p->bDistributeMotionEstimation && !p->lookaheadSlices)
        allowPools = false;

	//根据核数检测线程的数目
    if (!p->frameNumThreads)
    {
        // auto-detect frame threads
        int cpuCount = ThreadPool::getCpuCount();
        if (!p->bEnableWavefront)
            p->frameNumThreads = X265_MIN3(cpuCount, (rows + 1) / 2, X265_MAX_FRAME_THREADS);
        else if (cpuCount >= 32)
            p->frameNumThreads = (p->sourceHeight > 2000) ? 8 : 6; // dual-socket 10-core IvyBridge or higher
        else if (cpuCount >= 16)
            p->frameNumThreads = 5; // 8 HT cores, or dual socket
        else if (cpuCount >= 8)
            p->frameNumThreads = 3; // 4 HT cores
        else if (cpuCount >= 4)
            p->frameNumThreads = 2; // Dual or Quad core
        else
            p->frameNumThreads = 1;
    }

    m_numPools = 0;
    if (allowPools)
        m_threadPool = ThreadPool::allocThreadPools(p, m_numPools);

    if (!m_numPools)
    {
        // issue warnings if any of these features were requested
        if (p->bEnableWavefront)
            x265_log(p, X265_LOG_WARNING, "No thread pool allocated, --wpp disabled\n");
        if (p->bDistributeMotionEstimation)
            x265_log(p, X265_LOG_WARNING, "No thread pool allocated, --pme disabled\n");
        if (p->bDistributeModeAnalysis)
            x265_log(p, X265_LOG_WARNING, "No thread pool allocated, --pmode disabled\n");
        if (p->lookaheadSlices)
            x265_log(p, X265_LOG_WARNING, "No thread pool allocated, --lookahead-slices disabled\n");

        // disable all pool features if the thread pool is disabled or unusable.
        p->bEnableWavefront = p->bDistributeModeAnalysis = p->bDistributeMotionEstimation = p->lookaheadSlices = 0;
    }

    if (!p->bEnableWavefront && p->rc.vbvBufferSize)
    {
        x265_log(p, X265_LOG_ERROR, "VBV requires wavefront parallelism\n");
        m_aborted = true;
    }

    char buf[128];
    int len = 0;
    if (p->bEnableWavefront)
        len += sprintf(buf + len, "wpp(%d rows)", rows);
    if (p->bDistributeModeAnalysis)
        len += sprintf(buf + len, "%spmode", len ? "+" : "");
    if (p->bDistributeMotionEstimation)
        len += sprintf(buf + len, "%spme ", len ? "+" : "");
    if (!len)
        strcpy(buf, "none");

    x265_log(p, X265_LOG_INFO, "frame threads / pool features       : %d / %s\n", p->frameNumThreads, buf);

    for (int i = 0; i < m_param->frameNumThreads; i++)
    {
        m_frameEncoder[i] = new FrameEncoder;
        m_frameEncoder[i]->m_nalList.m_annexB = !!m_param->bAnnexB;
    }

    if (m_numPools)
    {
        for (int i = 0; i < m_param->frameNumThreads; i++)
        {
            int pool = i % m_numPools;
            m_frameEncoder[i]->m_pool = &m_threadPool[pool];
            m_frameEncoder[i]->m_jpId = m_threadPool[pool].m_numProviders++;
            m_threadPool[pool].m_jpTable[m_frameEncoder[i]->m_jpId] = m_frameEncoder[i];
        }
        for (int i = 0; i < m_numPools; i++)
            m_threadPool[i].start();
    }
    else
    {
        /* CU stats and noise-reduction buffers are indexed by jpId, so it cannot be left as -1 */
        for (int i = 0; i < m_param->frameNumThreads; i++)
            m_frameEncoder[i]->m_jpId = 0;
    }

    if (!m_scalingList.init())
    {
        x265_log(m_param, X265_LOG_ERROR, "Unable to allocate scaling list arrays\n");
        m_aborted = true;
    }
    else if (!m_param->scalingLists || !strcmp(m_param->scalingLists, "off"))
        m_scalingList.m_bEnabled = false;
    else if (!strcmp(m_param->scalingLists, "default"))
        m_scalingList.setDefaultScalingList();
    else if (m_scalingList.parseScalingList(m_param->scalingLists))
        m_aborted = true;
    m_scalingList.setupQuantMatrices();

    m_lookahead = new Lookahead(m_param, m_threadPool);
    if (m_numPools)
    {
        m_lookahead->m_jpId = m_threadPool[0].m_numProviders++;
        m_threadPool[0].m_jpTable[m_lookahead->m_jpId] = m_lookahead;
    }

    m_dpb = new DPB(m_param);
    m_rateControl = new RateControl(*m_param);

    initVPS(&m_vps);
    initSPS(&m_sps);
    initPPS(&m_pps);

    int numRows = (m_param->sourceHeight + g_maxCUSize - 1) / g_maxCUSize;
    int numCols = (m_param->sourceWidth  + g_maxCUSize - 1) / g_maxCUSize;
    for (int i = 0; i < m_param->frameNumThreads; i++)
    {
        if (!m_frameEncoder[i]->init(this, numRows, numCols))
        {
            x265_log(m_param, X265_LOG_ERROR, "Unable to initialize frame encoder, aborting\n");
            m_aborted = true;
        }
    }

    for (int i = 0; i < m_param->frameNumThreads; i++)
    {
        m_frameEncoder[i]->start();
        m_frameEncoder[i]->m_done.wait(); /* wait for thread to initialize */	//========调用threadMain()
    }

    if (m_param->bEmitHRDSEI)
        m_rateControl->initHRD(m_sps);
    if (!m_rateControl->init(m_sps))
        m_aborted = true;
    if (!m_lookahead->create())
        m_aborted = true;

    if (m_param->analysisMode)
    {
        const char* name = m_param->analysisFileName;
        if (!name)
            name = defaultAnalysisFileName;
        const char* mode = m_param->analysisMode == X265_ANALYSIS_LOAD ? "rb" : "wb";
        m_analysisFile = fopen(name, mode);
        if (!m_analysisFile)
        {
            x265_log(NULL, X265_LOG_ERROR, "Analysis load/save: failed to open file %s\n", name);
            m_aborted = true;
        }
    }

    m_bZeroLatency = !m_param->bframes && !m_param->lookaheadDepth && m_param->frameNumThreads == 1;

    m_aborted |= parseLambdaFile(m_param);

    m_encodeStartTime = x265_mdate();

    m_nalList.m_annexB = !!m_param->bAnnexB;
}

4、threadMain()函数的分析

        threadMain()函数相当于线程函数的main()函数，其主要功能就是在完成线程触发后等待处理，在此处，调用了compressFrame()函数。

        对应的代码分析如下：

void FrameEncoder::threadMain()
{
    THREAD_NAME("Frame", m_jpId);

    if (m_pool)	//若线程池不为空
    {
        m_pool->setCurrentThreadAffinity();	//设置当前线程

        /* the first FE on each NUMA node is responsible for allocating thread
         * local data for all worker threads in that pool. If WPP is disabled, then
         * each FE also needs a TLD instance */
        if (!m_jpId)
        {
            int numTLD = m_pool->m_numWorkers;
            if (!m_param->bEnableWavefront)
                numTLD += m_pool->m_numProviders;

            m_tld = new ThreadLocalData[numTLD];
            for (int i = 0; i < numTLD; i++)
            {
                m_tld[i].analysis.initSearch(*m_param, m_top->m_scalingList);
                m_tld[i].analysis.create(m_tld);
            }

            for (int i = 0; i < m_pool->m_numProviders; i++)
            {
                if (m_pool->m_jpTable[i]->m_isFrameEncoder) /* ugh; over-allocation and other issues here */
                {
                    FrameEncoder *peer = dynamic_cast<FrameEncoder*>(m_pool->m_jpTable[i]);
                    peer->m_tld = m_tld;
                }
            }
        }

        if (m_param->bEnableWavefront)
            m_localTldIdx = -1; // cause exception if used
        else
            m_localTldIdx = m_pool->m_numWorkers + m_jpId;
    }
    else	//若线程池为空，则WPP
    {
        m_tld = new ThreadLocalData;
        m_tld->analysis.initSearch(*m_param, m_top->m_scalingList);
        m_tld->analysis.create(NULL);
        m_localTldIdx = 0;
    }

    m_done.trigger();     /* signal that thread is initialized */ //线程已经触发
    m_enable.wait();      /* Encoder::encode() triggers this event */	//等待处理

    while (m_threadActive)
    {
        compressFrame();	//=====================调用compressFrame()
        m_done.trigger(); /* FrameEncoder::getEncodedPicture() blocks for this event */
        m_enable.wait();
    }
}