摘要： 随着深度学习在全球的风靡，算法模型层出不穷，如何将算法落地到生产环境中成为了热门研究领域。目前提高算法运行效率的主要聚焦点为优化模型结构、将权重数据进行量化，图优化等方面，然而，它们或多或少都会对模型精度带来一定的损失，不能做到完全的无损优化。

阿里妹导读：随着深度学习在全球的风靡，算法模型层出不穷，如何将算法落地到生产环境中成为了热门研究领域。目前提高算法运行效率的主要聚焦点为优化模型结构、将权重数据进行量化，图优化等方面，然而，它们或多或少都会对模型精度带来一定的损失，不能做到完全的无损优化。

作为工程开发人员我们能否从模型的执行模式上面进行相应的改造优化？流水线操作的优秀表现已经在工业领域得到体现，在不增加算法开发复杂度的前提下能否把它应用到AI任务调度系统中？基于这些问题，我们的流式编程框架应运而生。

一、背景

进入互联网下半场，人口红利逐渐削弱，新零售成为获取线下流量的重要入口之一。阿里搜索团队在新零售领域也进行了积极的探索，成立了客流项目，将线下商超进行数字化升级，商家根据消费者意图指定相关的改进和升级，提升购物体验。下图给出了线下门店实际场景的示例数据，从图中可以看到，实际商超环境比较复杂，人员分布随着环境不同也有很大的差异，在复杂环境下需要达到很好的数据提取效果，对深度学习模型的精度要求比较高，模型计算量很大，对硬件资源要求较高。

客流项目的整体软件工程模块主要组成部分如下图：

生产环境下，摄像头分辨率基本都是1080*1920以上，摄像头个数也比较多（某些门店高达30个以上），所部署的都是大计算量高精度的深度AI模型（比如行人检测，行人重识别和动作识别等网络模型），线下场景所用的机器GPU为GTX1060，这个配置的机器要求每秒处理这么多数据显然很有挑战，我们工程团队在如何高效进行AI计算方面进行了深入的研究和探索，提出了模型的流水线运行优化方案。通过如下图可以清楚解释流水线任务调度的优越性：

上图采用的五级流水线进行说明，可以清楚看到流水线模式使得同一时间从执行1条指令提升到5条，也就意味处理速度提高了5倍。正由于流水线在性能提升上有惊人的效果，我们将流水线技术引入到深度学习的工程化优化中，将硬件资源得到更大程度的利用，充分发挥CPU/GPU的计算潜能，并取得优秀的结果。也许还有人会疑惑为什么不采用传统的并发方法来实现呢？比如总共有三路视频，分别用三个进程进行处理，每个进程将所有的任务都做完，这样也能达到并发的效果。之所以不用这种简单并发方式原因是这样的：

1、模型任务之间是有相互依赖的，如果用简单多进程进行并发，执行到中间需要等待其他进程的数据，这样会造成资源浪费，比如在跨摄像头跟踪的时候，需要将所有摄像头数据进行分析，这就会导致一个进程的任务需要等待其他进程将其他路摄像头数据处理完之后进行统一调度分析，这种进程间的依赖和相互等待将严重降低运行效率。

2、 一台机器上的资源是有限的，多个进程都会独占存储资源和计算资源，多个进程进行单独运算意味着需要开启多个Tensorflow实例，会占用大量资源，一旦内存资源不够，触发了虚拟内存交换，就会很卡，导致运行效率急剧下降，甚至会出现频繁crash的情况。

3、计算资源在多个进程之间切换会大大降低计算效率。

4、流水线技术对执行任务进行了切割和重排，使得有限的资源能够发挥最大的效用。

由于简单并行具有很多局限，我们将工程代码进行了流水线模式改造，AI任务的流水线运行模式的实时发布，使项目性能得到了大幅度的提升，从之前的最多只支持实时处理8路摄像头数据提高到了16路摄像头，计算效率提高了一倍，资源利用率也得到了大幅度提升，最大化利用了硬件资源，在相同硬件资源下，能处理的数据量翻倍。

当然这个改造过程也是有成本的，我们在享受流水线技术带来的计算效率提升的同时，也带来了代码复杂度的指数级增加。在实际的生产过程中人工流水线的主要缺点有：

模型迭代效率低：新增模型功能需要很大的改造成本，基本需要将算法之前写的代码重构，融入流水线代码中；

任务调度与算法逻辑相互耦合：给人感觉从高级语言进入汇编语言，代码的晦涩难懂程度可想而知；

灵活性差：对任务的描述能力不强，任务调度不够灵活，抽象程度低，很难将一些新的优化措施应用上去。

这些缺陷严重制约了优化方案的应用，怎么样才能既能够提高计算效率又不影响模型的迭代效率？最好还能让算法同学容易上手，在这种情况下任务自动流水线化框架应运而生。

二、人工构造流水线与自动流水线主要对比

人工流水线：指的是需要人工将算法过程进行相应的分割，并人肉将任务用独立进程来运行，数据通道需要高度定制化。

自动流水线：指的是所有的任务调度过程，数据通道构建全部自动化完成，只需要把训练模型时候的代码进行移植过来即可，无需关心流水线任务调度所需要的中间过程。

我们分别从算法逻辑与流水线调度逻辑关联性、数据管道创建、开发灵活性、稳定性等几个方面进行了对比，从下面表格看出，在上个版本人工构建流水线的开发体验是相当的糟糕，各个方面都是完败自动流水线框架的表现。

三、系统框架介绍

1、设计理念

开发这么一套任务自动流水线化框架之初，我们对采用哪种编程语言、如何让算法熟练简单高效描述算法任务进行了多方面的讨论和论证，基于这个框架的目标客户是深度模型的算法同学，所有的尝试和优化点都会基于以下两个理念进行：

以模型训练完代码直接复用为第一要义：模型训练完之后代码可以直接移植到生产环境运行。

以服务算法同学为第一准则：模型任务描述简单高效容易上手。

鉴于大部分深度学习的算法同学在训练模型的时候都是基于Python语言编码，他们使用的训练框架是大众化的Tensorflow，经过与算法同学探讨后，我们果断选用Python语言进行开发，可以更大程度提高算法模型的迭代效率，对AI任务的描述也是参考Tensorflow构建模型的过程，这样可以大大简化算法同学的开发难度和学习成本，容易上手。

2、系统启动流程

这里借用了Tensorflow对模型的强大描述能力，参考Tensorflow框架的的构建和运行流程，自动流水线框架的启动运行流程如下：

从框架的运行流程图可以很直观的看出，主要分为以下三个阶段：任务描述、构建任务关系图、图运行，这些过程都能够类比到Tensorflow的启动过程，从框架运行上，会使用Tensorflow的算法同学肯定会很方便就能上手使用自动流水线任务调度框架。

3、主要模块

根据工厂任务调度和监控的完备性，我们对流水线运行时期的各个阶段进行了高度的抽象，其中各过程模块跟实际工厂车间运行机制高度统一，方便框架的理解和抽象，研发了一整套任务自动调度和监控的数据工厂体系，借助IPC跨进程通讯机制和SHA共享内存技术，完成流水线之间数据传递和通讯的需求，整个工厂的生命周期内，需要负责将上层输入的巨量图片数据自动进行高效加工，生产出数据特征传回服务端供商家使用。

整个工厂框架涉及到的模块主要有：

任务解析器：负责将算法描述的任务流图进行解析，完成数据流动管道的梳理以及任务与管道的绑定。

任务调度管理器：主要负责流水线上任务的调度管理工作（例如：将更多资源调度给管道比较拥挤的任务、没有数据的时候进行休眠）。

容灾恢复：当流水线在加工数据的过程中遇到异常退出的时候，创建新的进程继续工作，保证流水线数据的正常流转。

任务监控器：负责监控流水线任务的执行情况，如有死循环或者异常，能够及时发现。

跨进程数据管道：负责将数据从流水线传递到另一条流水线。

通过以上主要模块，我们可以顺利搭建好工厂的运行环境，并能够处理基本的异常恢复和监控，进而保证工厂稳定良好的运行环境。

四、流式编程

算法需要对AI任务进行分割（比如前处理、inference、后处理），包装成独立的计算模块，然后使用自动流水线框架API描述任务的输入输出接口的数据结构、任务间的依赖关系，跨进程通道的数据结构，参考工厂的组建过程代码如下图所示：

描述工厂构造过程的主要模块如下图所示，开发描述过程主要包含了AI任务的输入、输出数据结构的定义，执行函数名（指向指向函数体）。数据结构支持树形数据结构定义。工厂里面的解析器负责将各个任务关系进行解析，根据输入输出数据结构定义，通过跨进程内存管理器构建数据流管道。通过任务间依赖关系构建任务数据流向关系，为了提升内存利用效率，我们专门为图片设计了多流水线共享管道，并构建图片索引队列，任务只需要根据图片索引从共享管道获取需要处理的图片，省掉了图片数据的流动，大大提升图片的存储耗时同时也节省内存。利用执行函数名构建执行任务实体，并塞入到执行进程的任务列表中。

五、任务调度系统

CPU 是多核时是支持多个线程同时执行。但在 Python 环境中，无论CPU是多少核，Python的解释器同时只能由一个线程获取。其根源是 GIL 的存在。GIL就是一把巨大的锁，线程执行的时候需要获取这把锁才能获取解释器，同一个进程只有一把GIL锁，拿不到这个锁的线程只能等待其他线程把GIL锁释放后再执行，因此Python中的多线程并发都是逻辑上并发物理上串行执行，有兴趣的同学可以自行了解GIL的前程往事。

由于GIL锁的存在用多线程实现AI Inference子任务并发提高效率的愿望落空之后，只能选择多进程利用CPU的多核实现高并发提高数据吞吐量，在多进程开发中，每条流水线中的任务需要满足百分之九十以上时间只会使用同一类资源（例如只需要CPU或者GPU）,如果达不到这个条件就没办法最大限度利用硬件资源。

1、任务流水式执行过程

假如有十个任务，四条流水线为例进行说明，任务以及数据传递通道如下图所示，各个任务的耗时以及所需要的CPU/GPU资源不一样，在部署任务的过程中会考虑到资源和耗时都需要保持相对平衡，然后将各个任务部署在不同的流水线上，任务部署后数据的流向看起来会毫无规则，实际上入流水一般没有阻碍。各个任务从理论逻辑上面看是串行运行，但是从各个任务的执行实际和调度机理上看各个任务都是独立并发执行。

每条流水线任务的执行受到独立任务调度管理器控制，在循环数据加工过程中，会存在某些输入管道或者输出管道比较拥堵、某些任务需要批量跑数据但是任务的管道容量或者数据不足以容纳这么多Batch的数据，当然还有其他一些条件就不一一列举，这时候会优先调度其他任务，任务经过数据加工厂的高度抽象化，可以很方便的进行数据任务的调度优化，进一步提升数据加工效率。

2、流水线通讯机制

从上面的流水线执行逻辑图上可以看出，各个流水线的输入数据和输出数据具有不确定性，这就导致了一个问题：为了节约硬件资源，当流水线的所有输入管道都没有数据的时候流水线会进入自动休眠状态，但是谁来负责将流水线唤醒？

毫无疑问，首先想到的是用信号量，以流水线1为例说明，流水线1有三条输入管道，分别为上层输入的元数据、流水线2上面Task7的输出管道、流水线3上Task9的输出管道，也就是说流水线需要等待三个独立的信号量，信号量有个问题就是只要一个信号量获取不到当前流水线就会被休眠，这势必会堵住当前流水线工作，流水线需要的工作机制是只要有一个管道有数据就会工作，只有所有管道都没数据的时候才会休眠，这只是考虑到三个管道的情况，还没有考虑输出管道都满了的时候，如果都考虑的话甚至会存在多达几十个信号量，这极大增加了流水线工作的复杂度和效率，是万万不能接受的。

我们舍弃了用信号量的方式控制流水线的工作状态，以数据共享管道为介质，让管道主动唤醒休眠状态的流水线。在工程初始化的时候，会分析出数据和任务的关系图、数据流转图，通过这两个图将相关联的流水线绑定到各自管道上，只要管道有数据输入就会唤醒下游的流水线，通过这种方式就很好的解决了信号量爆炸的问题。

六、高性能跨进程共享管道

在多进程实现的流水线方案中，由于每个进程的数据都是相互独立的，一个进程产生或修改的数据对其他进程而言它是无感知。如何提高进程间的数据传递是能否高效实现并发的关键点。

在Python中实现多进程共享内存的现有解决方案主要有：管道、multiprocessing.Queue,multiprocessing.Value,multiprocessing.Array,前两个主要是利用Pipe实现的管道在内核分配一个缓冲区并进行管理，后两个主要是利用mmap来存储ctypes类型的数据结构，后两个的数据存储性能较前两个高很多，缺点也比较明显，Value/Array只支持对Ctypes基本数据类型的存储，对复杂的数据对象类型不支持，我们的自动流水线方案中需要存储自定义的类结构，显然达不到性能高灵活性高的两高要求。既然现有的方案不能满足我们的需求，需要我们自己开发支持复杂对象的高性能跨进程通道。

1、管道主要模块

从自定义对象到存储底层，主要设计了如下图所示的几个功能模块，跨进程共享内存使用了SHA技术作为存储底层逻辑，SHA相比mmap各有优势，两者的区别和优劣网上有很多文章进行详细的讲解，我们选择SHA的一个主要原因是它比较容易就能够支持动态扩展存储节点，比较方便的支持了List数据结构。

从一个自定义对象数据到存储在SHA共享内存，中间主要经历了对象结构的分拆、数据序列化、数据维度和类型的处理、内存块拷贝，读操作跟这个过程的逆操作，读和写的主要区别在于写数据会存在一个数据拷贝的过程，读数据将这个过程优化掉了，直接将共享内存的数据格式转换为目标数据，读数据的性能得到了大大的提高。

数据分拆过程主要是对类数据结构进行递归解析，将数据结构的属性分拆为最基本的数据类型，为每个属性分配了一个属性队列方便数据操作。利用反射机制高效进行数据序列化和反序列化。

2、数据序列化和反序列化

自定义数据结构的序列化和反序列化操作如下图所示，主要有两个过程：

类对象 ↔ 数字序列，可以将类结构组织为树状结构，非叶子节点表示为自定义数据结构，叶子节点表示由基本数据类型（float/int/…）组成的数字序列。

数字序列 ↔ Ctypes数据，主要将numpy的数据结构转换为ctypes的类型，并将数据维度进行归一化/反归一化。

使用python的反射特性，通过传入的数据结构类型信息解析出树形节点，并转换为ctypes的数据类型，以这样的数组作为数据管道的存储元素，构建通讯管道。

七、流水线容灾恢复

作为一个长期运行的工厂体系，能够持续稳定高效率地将元数据加工为产品是我们的追求目标之一，提高系统质量加强代码鲁棒性减少bug或者异常是基础工作，但是一旦遇到异常能够恢复也是必不可少的部分，我们还是有必要引入基本的异常恢复功能模块。

因为Python语言GIL锁的原因我们的多流水线对应的是多进程，只有在多进程下才能享受高并发特性。多进程带来高并发的同时也带来了高复杂度，流水线之间的工作相对独立，一个进程发生异常其他进程是不会知道的，其他进程依然会运行，但是中间流水线中断工作后，导致整条数据管道中断，工厂将无法工作，为此，我们专门做了多流水线监控机制，如下图所示：子线程会每隔一段时间检测流水线状态，如果发现有某条流水线发生异常，则启动异常恢复机制，更换流水线进程让这条流水线死灰复燃。

流水线能够恢复的一个前提是进程发生异常的时候需要将现场数据进行存储，待流水线启动异常恢复更换进程，下一个进程再读取上一次现场数据继续往下加工，这个控制流程如下图所示：

八、性能数据

1、跨进程共享管道性能测试

在CPU(Intel Core I5-45903.3GHz4)的机器上，测试读取（10801920*3）的图片一千次，分别测试了基于Pipe管道的Queue队列、基于mmap存储Array、基于SHA自定义的管道三种存储方案，测试性能如下表：

通过测试发现，我们实行的基于SHA支持复杂数据结构的管道读写数据性能都很接近Python自带的Array方案，但是Array方案的缺点就是不支持序列化复杂对象，只支持基本数据类型。相比同样支持序列化的Queue队列，我们自己实现的管道在读取耗时上比Queue快了400倍，在数据写入上因为Queue采用的是异步写入，所以测试出的数据不能真实反映真正存储到共享内存锁需要的时间，相信真正写入的耗时远远大于测试出的数据。

2、自动流水线整体框架性能

我们分别对串行版本、线上的人工流水线并行版本、当前的流水线任务自动调度版本性能进行了极限测试对比，测试环境如下：

测试结果对比如下：

从实际测量数据来看，人工流水线版本带来的性能提升是完全符合预期的：支持视频的路数多了一倍，但是手动构建流水线复杂度高，影响算法的迭代效率。

受益于自动流水线框架的高度抽象，很多任务调度优化能够很方便的应用到流水线上面，从数据测试上也能看出这个优势，自动流水线框架性能在人工流水线版本之上得到进一步提升，从16路摄像头提升到了18路摄像头，单帧AI耗时从13.7毫秒降低到11.9毫秒，性能提升13%，硬件资源的使用率也是符合设计预期的。

自动流水线版本对比人工流水线版本数据提升如下：

CPU的使用率提升了20%

GPU的使用率提升了10%

摄像头数提升12.5%

单帧AI耗时降低13%

通过自动流水线框架对AI任务的运行流程进行描述，开发复杂度得到了很大程度的降低，编程模式得到了算法同学的认可，模型的运行效率得到大大提升，模型迭代效率得到了质的提升。

九、项目业务效果

目前我们客流数字化项目已经上线运行有较长一段时间，已经有很多商家接入了我们的客流系统，所采集到的特征数据已经在相应的产品中得到应用，一些商家基于我们分析出的数据对商超布局、商品摆放顺序以及商品种类进行了相应调整，购物体验和交易量都得到了相应提升。

我们分别以区域热力图和区域动线图来说明我们的业务展示效果，区域热力图和区域动线图分别展示如下：

区域热力图的计算方式为：将门店的所有摄像头数据进行整合分析，最终将反应人流密度的数据映射到门店的平面CAD图上形成区域热力，这主要向商家展示在一天内门店各个区域各时段的人流密度，让商家能够清楚直观看到各个区域的人流热度。区域动线图展示了各个区域的人流以及人流的流向，基于这两个数据图，商家能够清晰了解到门店各个区域的人流密度情况以及消费者在门店的流转状态，通过人流动线图可以分析出商品的摆放是否合理，消费者是否比较方便选到自己中意的商品，有了这些数据，商家在调整销售策略的时候便有了数据依据和方向。

总结和展望

本文主要介绍了自动流水线任务调度系统在客流项目中的应用，相信流水线的编程思想在工程实践中优化性能上还有很大的利用价值，相信很多工业界成熟优秀的运行模式都可以运用到软件工程中，对我们以后在运行效率提升方面的探索研究具有很强的指导意义。

自动任务调度系统还可以应用到更多其他项目中，在训练算法模型阶段也可以尝试接入，提高训练效率降低训练时间。自动流水线框架的目标用户是算法同学，提高算法模型的迭代效率是我们的第一准则，在保证性能的前提下尽可能的方便简单。

尽管从上面的测试数据看，硬件资源以及得到了极大程度的利用，但是依然相信还有很多可以挖掘提升的地方，模型运行性能和稳定性都能够进一步提升。更多的商家和业务在接入中，我们在不断优化技术的同时，也在和行业商家一起不断打磨产品体验和细节，争取让客流项目更好地赋能商家赋能新零售。

作者：技术小能手