晓兵技术杂谈4-分布式存储系统-daos的异步任务调度引擎框架TSE_task_schedule_e原创

Hello，大家好，我是小兵，这是我的第四个技术视频，今天给大家带来的分享是大S的异步任务调度引擎，希望大家喜欢。哈喽，大家好，我是小兵。今天给大家带来。分布式存储系统，Dells的异步任务调度引擎。也就是T1。Task schedule engine。我们先大致看一下流程啊。有时候。在这个分布式存储系统中，我们。首先分为那个。两个。主要的结构。

TC主要分为两个主要的结构，一个是任务结构。也就是我们的task。但这个task上面有一个primary。呃。就是他的私有属性也是dtp。还有一个它的调度器，调度器的私有属性DSP。我们的任务上面。可以挂一些，呃。依赖任务的对立。还有那个欲回调的，对你。完善回调对。呃，批量执行任务的列表可以是单独的放在这里。啊，当然这个任务也有一些的状态，完成状态或者是运行中。然后这个调度器上面放了初始对立。很完整，对的，运行时对的。休眠对立。一般用于延迟任务的队列。还有这个完成对应，当然这些我们可以基于这个来扩展，因为这个TC模块它是独立的一个。

项目的模块。我们先大致看一下流程。比如。当我们。创建一个任务的时候。首先他会注册。在创建任务中注册。任务的一个回调。我们调用task cb。也就是说他把这个我一定要给注册到那个DTCCB上面。嗯，也就是说他把这个任务。的回料加到了。Dtp。也就是。为插插到这个。完成回调队列上面他的这个任务的完成回调上面。然后紧接着他往下。

创建完任务，我们就会调度一个任务，调度任务的时候。这边它分为那个。默认师傅延迟调度就是立即调度task schedule delay。然后这边判断任务是那个ready。无依赖，并且是无那个预回调任务的函数，就这队列里面已经被执行完了，或者说他没有依赖任务。那么就立即执行。然后。如果效力执行的话，他就是。把那个任务放到那个运行队列里面，然后这边往下走。如果是非延迟的调度任务的话。就把它加到那个初始化任务的列表里面。就是待执行。其他的条件的话就是说。

需要等那个唤醒时间，如果满足那个唤醒时间。也就说我预定一个唤醒时间，然后判断当天时间加上这个延时是否是和这个唤醒时间相等，相等的话，我们就把那个。嗯，加到那个唤醒队里。然后当我们执行调度的时候。调度的时候就会走在下面来。呃。他调度这个地方，它有一个入口，就是说。我们在调度的时候，我们要去那个。出发回回调在这边。是那个。Cut的。那个带条件的那个去吹，就是去拿那个网络层的一些事件，然后来触发这个回调。就触发这个。

时间的毁掉。那事件的回调的，它这边它里面会推进那个任务的执行。然后执行任务的调度。最终是找到这下面来。他会去便利那个，消灭对立。判断那个是否是。到了。唤醒时间，如果未到唤醒时间的话。入微的话，你时间的话就把它加到那个。呃，初始的表。但这个休眠队。就从这里拿这里加进去的，然后从这里面去便利的。然后他这个执行任务的时候，就先便利睡眠队列，然后再便利那个初始队列。所以这里如果没有依赖或者任务被取消。那么就。把它放到那个，呃，没有依赖，或者任务被取消，放到那个临时的一个队列list里面。然后紧接着又会去编辑这个list，然后如果认为是嗯。

如果这个任务是没有被取消，那么就把它放到那个运行队列里面。紧接着它会调一下那个主体的任务函数。我就是我们过的时候。主体好像一般都会去走网络，或者他就是走RPC。走网络。别找到这边渗了，渗了之后它又会掉这个，呃，发送回调就发送完了之后，然后又走到这个发送回调里面去执行这个任务完成。我执行任务完成的回调，然后去便利那个任务就是我们最开始注册的那个。呃，任务完成，回调拿出来，把里面那个函数来执行一遍。这执行任务注册实时的毁掉，其实就是上面，比如说我们注册一个回调。就执行这个公共的compare的这个回调。然后这边可以是把任务用来完成。

完成的话就在这边加锁的完成，然后他会把那个移动到那个，呃。从把那个任务移动到那个。从调度器移动到它的。调度器的完成队列里面去。大概的一个过程就是这样的。然后我们。接下来看一下代码里面的。其实通过这个TS模块，我们。就可以对一个分布式存储系统的异步任务调度引擎。有一个非常详细的了解，它里面设计的非常灵活，比如说我们从它的接口入手。

我们从这个地方随便找一个哈，我们就找一个写的流程，比如说这里有个。创建一个任务。当我们创建任务的时候。

我们可以看到需要传入一个任务的一个主体函数。也就是这个方形，然后还有一个调度器。还有一个那个task新，也就是出餐，就是我们在外面。初始化的一个task的指针，Task类型的指针，然后把这个的地址传进去，它会在里面把这个task包括一对的初始化画好，然后当执行完这个，它是create DC客户端的创建完成之后。就可以拿到这个task，就可以从这个task上面拿它的参数。他通过嗯，他私有的一些内嵌的参数，通过偏移或内存计算，然后拿到它的参数，然后在这个参数上设置类型，比如说我们的那个oh，也就是他的handler。然后。然后还有比如他的ID，就是IO的描述，或者说他的SG，然后紧接着就会调那个调度，然后我们先看一下创建任务的时候。

里面就是对TS进行了一层包装。我们首先取得二，然后。我们看这它首先是判断这个调度器啊。因为我们这个EV是带下来的，EV是。相当于是一个内部支付，对网络上下文做了一个呃绑定。然后这个调度器我们传的是。空的。这个EV我说一下怎么来的，EV就是说是在这个地方。这个E又是另外一个概念。呃，就1V，他会和那个。呃，调度通过一呃时间队列关联起来。

就今天我们主要讲那个task，然后。我们看到这个EV上面它设置了回调。其他这种是把那个E外层EV，它还包了一层DES的内层的一个E维。比如说我们看到这是defff的UV，所以叫de，就de下划线什么什么。然后这里面是包了一层内层的那个英文特。De event-T这个类型，这就是我们刚才看到的那个回调函数。它在回调的函数的时候，传的参数就是也是这个它本身。DEP event。我们看一下这个event event的结构。但是英文的结构的话，就是说它有一个。嗯，私有的一个。书信。Private。

这个会和创建那个EQ事件队列绑定起来。当我们那个启动的时候，会创建EQ。好，我们回到刚才创建任务的地方。做一些任务的时候。就是如果你的调度器为空，我们刚才传的是空，然后。然后当你的EV。也等于空的时候。才会走这面我们1V是传进来的部位空，然后。会从那个EV上面把那个调度器拿到。调度器类型我们可以大概看一下。压滤器类型的话，它也是这样做了一个外层的封装。调度器类型，然后。主要是他这个调度器的这个私有属性上面。有很多的那个。调度器的队列。

待会我们会说到。然后这个调度器。创建任务的时候。也是我们传入一个function，就是上面业务的执行的任务主体函数，然后把这个调度器给传进去，然后再传一个私有的。数据。再把我们的那个task的地址再往下传。那这个时候我们。就可以看到它是内部，这是一种常用的用法，就是说你这里task是传到二级指针，他是外层的task的地址传进来，他在内部，其实重新先。初始化task。造一个这个它是个T的。这个task出来。我们看到这个。它上面有很多那个。属性。它也是一个私有数据段，然后待会我们看一下，然后他先把这个taskco给分配出来，也是类似于male克里面封装的ma。

然后。封出来之后。初始化这个。DSPDSP就是task上面的那个私有的数据结构。通过一种方式把它拿到。就是这个私有数据的地址，然后转成那个呃，私有的那个结构，我们主要看这个私有结构，这个任务的私有结构上面，也就是我刚才那个图里面提到的上面，比如说有调度器，有呃任务的主体函数。还有他的一个链表。这个列表对。这个链表是用来练到那个。呃。练到调度器上面的。我们看到这里是第。它是dtp，也就是。

这个task pri，还有一个DSP。那就是。Private，然后这边有任务的wake up时间。嗯，有他的依赖任务的一个列表。还有一个预准备。也就是说，在执行任务前需要执行的回调，然后任务完成。回调任务的。状态完成运行中。然后下面有一些字段。他的一些。私有属性内部的。还有他的那个站。还有它内嵌的一个站的顶部位置。然后。的一个生成的一个标识。它的这个版本也是可以用来传参数的。然后我们再往下。

切到刚才那个地方。当他拿到这个四个属性之后，他就把这些列表。他的列表给初始化，它的依赖任务初始化，然后它的完成回调。它预。他任务执行全都毁掉，这些列表都初始化应用技术，然后。一。那我们这边。也是说他会。所以就是说任务的二级指针传进来，拿到第二度私有数据，新建一个任务，测试该任务的列表。依赖队列完成回调队列任务，前置回调函数队列任务引用技术加一设置主体函数私有的参数和调度去返回该任务，也是我们说的这里面的内容，然后返回就是体现在这个地方。他把二级指针取他的一级指针，然后把这个task给附上去，这样就把这个task给返回。

所以我们看到。创建完这个任务之后。这边紧接着他就是说。创建完任务，然后这边是任务任务，然后又把这个参数拿到，给他加了一个那个魔术字。用于辨别那个任务，然后如果有这个EV的话。会注册。我以为不为空。然后。给这个任务注册完成回调，那么这是一个公用的一个。完成回调。我们看一下这个。

这边的。Task还有。呃，完成回调。参数top为零。他是直接掉了租车，然后再往下。我们看到他把这个完整的回调主要是设到那个。呃。未完成回调就是加到那个dt task的完成汇调上面去了。要任务完成之后。出发在哪个回调？之后就会执行这个。所以外面这个task com的用task来完成这个。实践。

反正世界里面。主要是把这个事件给完成，实践呢，我们后面单独出一期节目。任务创建完成了。金色的它会调度。你看到这里他。任务创建完成了之后。把参数设置上之后，就会调度任务执行。调动任务执行，调动任务执行的时候。船队1TASK就立即调度。Instant就立即调度律师为处。然后。往下走。

他我们看他主要是看看往哪里，首先他把这个1V拿到，如果有1V的话，就把这个。意味启动。启动事件，把EV的标记为启动。比如说这个里面标记就会启动。熊世界。就已就就是已就绪的事件才能启动，也就是说他之前是调的那个初始化事件，它会这里会检查子事件的个数，如果。并种种加上已完成的子事件个数小于事件的总数，则返回所有的子事件完成之后才能启动负事件。然后如果事件正在终止中，那么就直接退出。这就是这个启动事件的里面，主要主要做了这些事情。它这个加速启动事件的话，就是将事件修改，修改状态机为运行中。如果有负时件，则为负事件的子事件命中技术器加一，它主要是管理那个负值依赖事件。如果有子时件在启动屏障。

之前副世界，之前完成则直接完成副世界。也就是说他的子事件都已经完成完了，然后这个时候传入的是复事件，就直接把那个复事件给完成。所以我们接下来看。因为。不等于read的话就直接返回，也说必须是就绪的事件才能启动。然后这边是判断那个指数线。微信中的事件将驶高速再返回。就说。运行中的加上已完成的子事件小于时间子时间的总数。您说他还有几十件没有完成。所以。就不能够把这个事件标为启动。然后。再往下。就是说他。如果这个他的那个。拿到那个事件上面的那个EV叉私有数据上面有一个EQ的控制器，他去查，查到了之后。

如果这个时间队列。时间队列在处于那个终止对终止的一个状态的话，那么直接返回。报错，否则的话就是说枷锁把这个事件给启动起来，也就是我刚才说的这里面的。然后这边也是。大家可以看到这个C语言的任务。比较标准的一种风格，开源。分布式存储系统，De的这个代码风格是非常好的。包括他的注释。他是日常出身。这值得我们借鉴和学习。我们回到刚才的地方。那提到外成。然后。

这个我们一般是返回零，所以这这下面的步骤。然后退出这里。他开始调度任务。真正的调度任务，那个任务。先打开看一下这里面做了什么。调度任务立即调度，首先就从task拿到调度器。说没有依赖任务前前置任务已经做完了，那么就可以。执行。感染物。可立即执行该做任务。这里面的进去看一下。你看它这是一个灵活，是一个精密的状态机。很instant。受理调这开始调度D内我们可以传一个延迟调度的时间，他尽量会做一些判断。

就首先我们判断ready的条件。就什么时候这个任务可以执行的话，就是说他的依赖任务为空，并且就是他的没有依赖的技术了，这个任务上面，并且说他的那个回调前置回调也为空，也就表示前置回调都已经完成的情况下，那么这个ready就为出，然后。但是这个是立即执行。或者说他的没有那个一会药。呃，没有那个任务，主体函数等于空。那么就会走下面。然后他主要是把那个加到运行队列里面做一把任务加到运行队列里面做一个管理，然后。这边枷锁。增加了用激素。就是如果那个。不满足这个条件，但是它的D类等于零，也就是说我们需要那个立即执行，立即执行的话，他就把它放到那个初始对立，初始对立的话就会立即，后面就会立即执行。

我们看到。如果这个它是这个ready为真的话，并且这个需要立即执行为帧的话，它这里会执行一下这个主体函数。DDP。也就执行我们的那个。这个这个主体函数。所以就掉了我们的主体函数。这就是任务的创建和执行。我看一下大概它提供了哪些口。TC调度器的初始化对。

还有调度器完成，可以把这个调度器上所有任务都完成。初始化。这些都是。创建任务之前都搞了这个初始化，然后再注册任务毁掉，然后。推进那个调度去执行，他检查是否完成啊。调度。调度完成，获取一些参数等，还有把那个任务用来做那个重重置。这就是对外暴露的接口。测试，我们可以也可以通过单元测试。看一下大家等级这个从。这个单元测试是用的框架啊，这边单元测试运行run的时候，执行这个编译完会有这个。单元测试从面进来，然后。

这边运行C卡主要是。这边是前置的。调度器前置，我们看调度器前置。因为主要是把那个第bug给初始化了，这个我们不关心。然后开始执行这个里面的单元测试，单元测试就是这个。数组里面有，呃，我我调了顺序，调试的时候把六放到一的前面，本来是一二三四五六七八九十。每个的测试我们看一下看。来测试。用这个关键词看一下。比如第一个测试，他就是测试创建完成和取消，它先是初始化这个调度器，然后创建。任务。然后调。然后调度。调度这个任务，然后检查这个调度器是否完成。然后他可以直接把这个任务给完成。

或者说检查这个调度器是否已经完全和上面是一样的。这边它又开始初始化，它是开始重置这个调度器。创建任务。大概我们看一下里面的接口。所以这个可以作为第三库引入到其他项目中，也可以。做一个非常好的一个借鉴参考。初始化调度器的时候，我们看到是在外面把这个。调度器传进来的，它初始化的时候。初始化的时候主要是。看你有没有传回调函数。没传为浪浪的话，就没有回调函数，没有回调函数这调度器主要是。把它调度器上面的初始列表，运行队列啊，完成队列啊，休眠队列，还有它的那个。呃。这里完成对列，这里是完成对列，完成回调队列，一个完成队列，一个完成回调队列。

然后对他的。完成。回调注册。出售调度器主要就是干这个。一个典型的一个队列的初始化。说的，我们刚才说其实这个已经说了，然后把任务给创建出来，然后开始调度。调度的话，我们刚才也说了。里面。会进行大家进行一个调度。然后我们检查这个调度器是否完成的话。就拿着这个调度器，把它的那个私有结构拿到。石油结构上面有锁。初始队列，完成队列，这些都在上面。我们可以扩展，然后他判断完成的条件，就是说。初始队里为空。并且。

休眠队列为空，并且它的飞行中的IO。计数器为零，那么我们就认为它完成，然后他就返回这个完成，但是这个可是处或者是force解决这里面的条件是否满足。这里flag如果已经完成的话。因为他这里面的任务没有没有主动调完成。所以这里面应该是有一个飞行的任务的，所以当他掉了这个，主动去把这个任务完成的时候。那再去检查的话。他的任务才任务接出去就会清零。我们看下他主动完成这个任务的时候。是怎么做的？拿了这个task。主要就是去掉任务的回调。执行任务的回调。去便利这个。Dtp。任务的回调的队列。

他去检查这两个任务是不是。呃，重入的主要就是通过最开始生成的那个。Generate保存了一个值，然后他去执行那个回调。那这个人物还支持依赖人物哈。所以它是非常灵活的。他执行完回调了之后，他去判断这个是否是看守，如果没有被看L的话。他这边去换成。丸子里面也是去。如果已经完成就直接返回，那没有。如果不是在运行中，那么就把这个。不是在运行中。就把这个计数器加一。然后运行中自为零完成自为一，把这个任务加到这个DSP调度器的完成队列。

再继续检查这个调度器，调度器肯定就完成了。这就是一个那个任务。调度引擎的主流程，但我们可以简单看一下里面。它可以测重入。好，完成回调。Win con。还有其他的任务。一二任务的时候就是说。

创建任务的创建依赖任务。这个接口。就注册在计划任务完成之前需要完成的依赖任务，如果依然无法进行，这如果就是依赖任务。只有依赖任务完成了，才可以将任务添加到调度机列表中，就是先完成依赖任务，比如说我们这个注册任务的时候，第一个传入的是一个。呃，一个主任务，然后后面这个任务，它依赖于这个任务，那么就要等这个任务执行完，才能够把这个任务加到运行队里，直接。他他回来的时候，他也会去检查这个依赖的一个情况。Progress里面。就一些调度。要把这个调度器运行起来。

这里面就有三个主要的大的步骤，就是初始化调度去处理完成的对立去。检查是否完成。这里面。有个地方会去处理。List。支持处理完成。我们看他处理那个依赖任务的时候，是在这个地方，是在调度。调度任务的时候，这里面如果你的。依赖任务不为空，然后就会去做依赖任务的处理。

依赖任务上的任务，队里不为空，说明主任务还有依赖任务。依然人不的话。这里面主要是通过一些计数器。去检查，或者说他判断是否是同一个调度器啊。然后他的一类任务是否是大于零啊。师傅们所有的依赖任务都执行完了。他就会去把。副什么加大那个运行队列之类的逻辑。饮用之类的，对。大家可以是根据这个单元测试去挨个去跑一下这个。从这里面可以学到很多C语言编程相关的技能。然。看一下它这个。

依赖任务它有一个关系啊。还有一对多或者一对一的一个关系。好，我们看到这里。对于依赖任务的一个结构，也就是说，一个探测可以是多对一。注册的时候。DOTA1的模型。这是我的一个主任务。他的那个计数器上面有三个子任务，那么它分别是这三个。它就依赖于这个，它是个零，它是个一，它是个二。因为每一个它这个它上面都有依赖的一个链表。

然后如果是一对多的话，多会依赖的话。做的方式不一样。我们看到这里。Task。我有多少个依赖啊，这边就是有一个依赖，但是他。是。这组任务有多个去依赖同一个task。你看这边主任务有。比如说有三个。那每一个他都去依赖后面的这个。每一个他会挂着这个。依赖任务的列表上面去。所以我们看到这个。Dest这个任务调度引擎。它非常灵活。考虑了我们。平时的所有。平时的大部分的编程的一个。情况。所以可以作为一个通用的一个调度框架来使用，我们可以基于它来做二次开发。

好了，今天的技术分享就到此为止。主要是对。DELL的任务调度。引擎框架做了一个大致的一个描述。欢迎。对这块。技术感兴趣的朋友。加入我们的dells技术交流群。好的。希望大家喜欢，谢谢，再见。