Ascend C是CANN针对算子开发场景推出的编程语言，原生支持C和C++标准规范，兼具开发效率和运行性能。使用Ascend C，开发者可以基于昇腾AI硬件，高效的实现自定义的创新算法。

目前已经有越来越多的开发者使用Ascend C，我们将通过几期“Ascend C算子性能优化”专题分享，围绕开发者最为关心的算子性能优化环节，介绍Ascend C算子常用的优化技巧，帮助开发者自主构建出更优性能的算子。专题内容将围绕流水优化、搬运优化、内存优化、API使用优化以及Tiling优化等优化技巧，从方案讲解、优化案例、性能对比等多角度展开介绍。下面进入第一期内容：Ascend C流水优化，您将了解到以下流水优化技巧：

• 基于Ascend C编程范式快速高效实现AI Core内流水并行
• 使能double buffer将待处理的数据一分为二，提高Vector单元利用效率
• 使能Iterate异步接口，避免AIC/AIV同步依赖 
  

基于Ascend C编程范式实现AI Core内流水并行

AI Core内部的执行单元异步并行地执行接收到的指令。每一个执行单元都可以看成是流水线上的节点，通过流水线并行的方式来提高计算效率。如下图所示，从输入数据到输出数据需要经过3个阶段任务的处理（T1、T2、T3），多个执行单元并行处理，每个执行单元只会专注于一个任务的处理，会处理所有的数据分片。

流水线并行示意图如下所示：

这里的流水线并行和工业生产中的流水线是类似的，执行单元1完成对某个数据分片的处理后，将其加入到通信队列，执行单元2空闲时就会从队列中取出数据继续处理；可以类比为生产流水线中的工人只完成某一项固定工序，完成后就交由下一项工序负责人继续处理。

基于Ascend C编程范式进行代码编写，实际上就是应用这种流水线式的编程范式，把算子核内的处理程序，分成多个流水任务，通过队列（Queue）完成任务间通信和同步，并通过统一的资源管理模块（Pipe）来统一管理内存、事件等资源。

Ascend C流水编程范式将单核算子处理逻辑划分为多个流水任务，CopyIn搬入，Compute计算，CopyOut搬出，基于该编程范式，可快速搭建算子实现的代码框架。以Vector编程范式为例：

• CopyIn负责搬入操作：将输入数据从Global Memory搬运到Local Memory（VECIN用于表达矢量计算搬入数据的存放位置），完成搬运后执行入队列操作；
• Compute负责矢量指令计算操作：完成队列出队后，从Local Memory获取数据并计算，计算完成后执行入队操作；
• CopyOut负责搬出操作：完成队列出队后，将计算结果从Local Memory（VECOUT用于表达矢量计算搬出数据的存放位置）搬运到GM。

从编程的角度来讲，Vector编程范式算子实现流程如下所示：

相关伪代码示例：

TPipe pipe;   //创建全局的资源管理    
TQue<VecIn, 1> queIn;  //创建CopyIn阶段的队列 
TQue<VecOut, 1> queOut; //创建CopyOut阶段的队列 
// Init 阶段： 
pipe.InitBuffer(queIn, 2, 1024);  // 开启double buffer,将待处理的数据一分为二,实现流水并行 
for-loop { 
    //CopyIn 阶段{ 
    auto tensor = queIn.AllocTensor<half>();     //从Que上申请资源, 长度1024 
    DataCopy(tensor, gm, len);                   //搬运数据从GM到VECIN 
    queIn.EnQue(tensor);  
    } 
    //Compute阶段{ 
    auto tensor = queIn.DeQue<half>(); 
    auto tensorOut = queOut.AllocTensor<half>(); 
    Abs(tensorOut, tensor, 1024); 
    queIn.FreeTensor(tensor); 
    queOut.EnQue(tensorOut); 
    } 
    //CopyOut 阶段{ 
    auto tensor = queOut.DeQue<half>(); 
    DataCopy(gmOut, tensor, 1024); 
    queOut.FreeTensor(tensor); 
    } 
}

按照上述编程范式进行编程即可实现单核上数据的并行处理。需要处理的数据被切分成n片，每个并行任务（Stage1、2、3）需要依次完成n个数据切片的处理。Progress1、2、3代表处理的数据分片，对于同一片数据，Stage1、Stage2、Stage3之间的处理具有依赖关系，需要串行处理；不同的数据切片，同一时间点，可以有多个任务在并行处理，由此达到任务并行、提升性能的目的。

流水任务运行示意图如下所示：

使能double buffer，提高Vector单元利用效率

执行于AI Core上的指令队列主要包括如下几类，Vector指令队列（V）、Cube指令队列（M）、Scalar指令队列（S）和搬运指令队列（MTE1/MTE2/MTE3）。不同指令队列间的相互独立性和可并行执行的特点，是double buffer优化机制的基石。

矢量计算前后的CopyIn、CopyOut过程使用搬运指令队列（MTE2/MTE3），Compute过程使用Vector指令队列（V），不同指令队列可并行执行，意味着CopyIn、CopyOut过程和Compute过程是可以并行的。如下图所示，考虑一个完整的数据搬运和计算过程，CopyIn过程将数据从Global Memory搬运到Local Memory，Vector计算单元完成compute计算后，经过CopyOut过程将计算结果搬回Global Memory。

数据搬运与Vector计算过程如下所示：

在此过程中，数据搬运与Vector计算串行执行，Vector计算单元无可避免存在资源闲置问题，假设CopyIn、Compute、CopyOut三阶段分别耗时相同均为t，则Vector的利用率仅为1/3，等待时间过长，Vector利用率严重不足。

未使能double buffer的流水图如下所示：

为减少Vector等待时间，可以使能double buffer机制将待处理的数据一分为二，比如Tensor1、Tensor2，如下图所示：

使能double buffer机制如下所示：

当Vector单元对Tensor1中数据进行Compute计算时，Tensor2数据流可以执行CopyIn的过程；而当Vector切换到计算Tensor2时，Tensor1数据流可以执行CopyOut的过程。由此，数据的进出搬运和Vector计算实现并行执行，Vector闲置问题得以有效缓解。

使能double buffer的流水图如下所示：

总体来说，double buffer是基于MTE指令队列与Vector指令队列的独立性和可并行性，通过将数据搬运与Vector计算并行执行以隐藏大部分的数据搬运时间，并降低Vector指令的等待时间，最终提高Vector单元的利用效率。通过为队列申请内存时设置内存块的个数为2，使能double buffer，实现数据并行，简单代码示例如下：

pipe.InitBuffer(inQueueX, 2, 256);

下面给出一个实际的使用示例，未使能double buffer：

__aicore__ inline void Init(__gm__ uint8_t* src0Gm, __gm__ uint8_t* src1Gm, __gm__ uint8_t* dstGm) 
{ 
    src0Global.SetGlobalBuffer((__gm__ half*)src0Gm); 
    src1Global.SetGlobalBuffer((__gm__ half*)src1Gm); 
    dstGlobal.SetGlobalBuffer((__gm__ half*)dstGm); 
    // 不使能double buffer,占用的物理空间是 1 * sizeSrc0 * sizeof(half) 
    // 3个InitBuffer执行后总空间为1 * (sizeSrc0 * sizeof(half) + sizeSrc1 * sizeof(half) + sizeDst0 * sizeof(half))  
    pipe.InitBuffer(inQueueSrc0, 1, sizeSrc0 * sizeof(half)); 
    pipe.InitBuffer(inQueueSrc1, 1, sizeSrc1 * sizeof(half)); 
    pipe.InitBuffer(outQueueDst, 1, sizeDst0 * sizeof(half)); 
    } 
__aicore__ inline void Process() 
{ 
    // 需要round*2次循环才能处理完数据 
    for (uint32_t index = 0; index < round * 2; ++index) { 
        CopyIn(index); 
        Compute(); 
        CopyOut(index); 
    } 
}

使能double buffer：

__aicore__ inline void Init(__gm__ uint8_t* src0Gm, __gm__ uint8_t* src1Gm, __gm__ uint8_t* dstGm) 
{ 
    src0Global.SetGlobalBuffer((__gm__ half*)src0Gm); 
    src1Global.SetGlobalBuffer((__gm__ half*)src1Gm); 
    dstGlobal.SetGlobalBuffer((__gm__ half*)dstGm); 
    // InitBuffer中使用2表示使能double buffer,占用的物理空间是 2 * sizeSrc0 * sizeof(half) 
    // 3个InitBuffer执行后总空间为2 * (sizeSrc0 * sizeof(half) + sizeSrc1 * sizeof(half) + sizeDst0 * sizeof(half))  
    pipe.InitBuffer(inQueueSrc0, 2, sizeSrc0 * sizeof(half)); 
    pipe.InitBuffer(inQueueSrc1, 2, sizeSrc1 * sizeof(half)); 
    pipe.InitBuffer(outQueueDst, 2, sizeDst0 * sizeof(half)); 
    } 
__aicore__ inline void Process() 
{ 
    // 开启double buffer的前提是循环次数 >= 2 
    for (uint32_t index = 0; index < round; ++index) { 
        CopyIn(index); 
        Compute(); 
        CopyOut(index); 
    } 
}

需要注意的是，多数情况下，采用double buffer能有效提升Vector的利用率，缩减算子执行时间。然而，double buffer机制缓解Vector闲置问题，并不代表它总能带来整体的性能提升。例如：

• 当数据搬运时间较短，而Vector计算时间显著较长时，由于数据搬运在整个计算过程中的时间占比较低，double buffer机制带来的性能收益会偏小。
• 当原始数据较小且Vector可一次性完成所有数据量的计算时，强行使用double buffer会降低Vector计算资源的利用率，最终效果可能适得其反。

因此，double buffer的使用需综合考虑Vector算力、数据量大小、搬运与计算时间占比等多种因素。

使能Iterate异步接口避免AIC/AIV同步依赖

同步模式指的是程序执行时，需要等待某个操作完成后才能继续执行下一步操作。 异步模式指的是程序执行时，不需要等待某个操作完成就可以继续执行下一步操作。

对于包含矩阵计算和矢量计算的MIX编程模式，调用Matmul Iterate或者IterateAll时，AIV（AI Vector核）发送消息到AIC（AI Cube核）启动Matmul计算。Matmul的Iterate和IterateAll接口提供了同步和异步两种模式。

为避免数据内存地址踩踏或时序错误等问题，可以使用接口的同步模式，编译时内部自动插入同步指令，但冗余的同步指令会降低算子的性能。若通过Iterate<sync=true>同步方式，每次调用都会触发一次消息发送，如下图所示（同步方式消息发送示意图）：

而通过Iterate<sync=false>异步方式，仅第一次需要发送消息，后续无需发送消息，从而减少Cube与Vector核间交互，减少核间通信开销。因此，mix场景推荐使用Iterate<false>或者IterateAll<false>异步接口，如下图所示（异步方式消息发送示意图）：

开发者可参考如下示例使能Iterate异步接口避免AIC/AIV的同步依赖：

TQueBind<TPosition::CO2, TPosition::VECIN>  qVecIn; 
TQueBind<TPosition::VECIN, TPosition::VECOUT>  qVecOut; 
mm.SetTensorA(gmA); 
mm.SetTensorB(gmB); 
mm.SetWorkspace(workspace, size);//其中，workspace为临时空间的物理地址，size为singleCoreM*singleCoreN大小的矩阵C占用的内存大小：singleCoreM*singleCoreN*sizeof(float) 
int16_t scalar = 2; 
 
while(mm.template Iterate<false>()){ 
    auto cInUB = qVecIn.AllocTensor<float>(); 
    mm.GetTensorC(cInUB); 
    qVecIn.EnQue(cInUB); 
    cInUB = qVecIn.Deque<float>(); 
    auto cOutUB = qVecOut.AllocTensor<float>(); 
    Muls(cOutUB, cInUB, scalar, baseM*baseN); 
    qVecIn.FreeTensor(cInUB); 
    ... 
}