关于workflow:Workflow的计算调度算法

Workflow 的计算调度算法

让大家好奇了这么久，明天终于写被 cue 到最多的话题：Workflow 的计算调度，包含独创的调度算法与相干数据结构。
P.S. 原文写于 2022 年 5 月份~

C++ Workflow 作为一个异步调度编程范式，对调度的拆解是有几个档次的：

• 用户代码层：提供工作流级别的结构化并发，包含串并联、DAG 和复合工作等，用于治理业务逻辑，组织要做的事件的依赖关系；
• 资源管理层：对不同资源外部做了协调和治理，比方网络、CPU、GPU、文件 IO 等，用起码的代价、做最高效、最通用的资源复用。

明天就重点介绍一下，Workflow 外部独创的计算调度算法，包含 Executor 模块（仅 200 行代码）及相干模块，整体是如何治理计算资源、协调不同计算工作，从而做到无论工作耗时长短，都能够尽可能平衡调度的最通用的计划。

而且看完之后，也能够对上一篇《一个逻辑齐备的线程池》中始终强调的生命周期有一个更好的了解。架构设计始终要强调每一个模块自身做到齐备和自洽，是因为更有利于演化出下层模块。

https://github.com/sogou/workflow/blob/master/src/kernel/Executor.cc

1. 计算调度面临的问题

无论是用何种计算硬件，计算调度要解决的问题是：

1. 充沛应用资源；
2. 不同类别的工作的资源分配；
3. 优先级治理；

第一点 很好了解，以 CPU 为例，只有来工作了就要尽量跑满 CPU 上的若干核。

第二点，不同类别是比方：每件事情由 3 种步骤 A->B->C 组成，消耗的计算资源是1:2:3，简略的做法是能够别离给予 3 个线程池，线程比例 1:2:3（假如 24 核的机器，咱们能够别离把 3 个线程池中的线程数配置为 4，8，12）。

但因为线上环境是复杂多变的如果消耗资源变成了 7:2:3，固定线程数计划显然不可取，不改变代码是难以匹配这样的情况。

这么做的另一个弊病是，如果一批提交 100 个 a，那么显然只有 4 个线程能够工作，难以做到“资源利用即用”；

还有没有解决办法呢？更简单地，能够引入动静监测耗时，然而引入任何简单计划都会有新的 overhead，绝大部分状况下这些都是节约。

持续看 第三点，优先级治理是比方：还是 A->B->C 三种工作。当初减少了一个 D，我想要尽快被调起来，简略的做法往往是给所有工作一个优先级编号，比方 1 -32。

但这并不是短暂的解决办法，编号是固定的总会往更高优的用完，而且工作本人都是贪婪的，只有有最高优先级，最终大家都会卷起来（不是

咱们须要的，是一个 灵便配置线程比例 、 充沛调度 CPU、且能够 偏心解决优先级 的计划。

2. 翻新的数据结构：多维调度队列

Workflow 外部简直所有的计划都是往通用了做，对于 CPU 计算，则是：全局一个线程池，和对立的治理形式。应用上，计算工作只须要带一个队列名，即可交给 Workflow 帮你做到最平衡的调度。

基本原理图如下：

Executor 外部，有一个线程池和一个根本的主队列。而每个工作自身，同时也是属于一个 ExecQueue，能够看到其构造是一组以名字辨别的子队列。

这种数据结构的组合，能够做到以下三点：

1. 首先，只有有闲暇计算线程可用，工作将实时调起，计算队列名不起作用。

2. 当计算线程无奈实时调起每个工作的时候，那么同一队列名下的工作将按 FIFO 的程序被调起，而队列与队列之间则是平等看待。

   例如，先间断提交 n 个队列名为 A 的工作，再间断提交 n 个队列名为 B 的工作。那么无论每个工作的 cpu 耗时别离是多少，也无论计算线程数多少，这两个队列将近偏向于同时执行结束。

3. 这个法则能够扩大到任意队列数量以及任意提交程序。

别离来看看算法是什么。

第一点：Executor 的线程不停从 Executor 外部的主队列中拿工作进去执行;

第二点：线程从主队列把工作取走、并筹备执行工作之前，也把工作从它本人的子队列里拿走。并且，如果该子队列前面还有工作，就把下一个工作进去，放到主队列中。

第三点：内部用户给 Workflow 提交工作的时候，Workflow 会先把工作按名字放到子队列。并且如果发现这是子队列中的第一个工作（阐明方才子队列是空的），便立即提交到主队列中。

算法自身相当简略，而提交工作时，只须要给调度器轻微的领导，既队列名(对应 Executor 的一个 ExecQueue)，无需指定优先级或计算工夫预估等信息。

当咱们收到的 A, B, C 工作数足够多而且数量相等，无论工作以什么程序达到，也无论每个 (留神是每个而不是每种) 工作的计算工夫多少，A, B, C 三个子队列将同时计算实现。

而主队列长度，永远不超过子队列的个数，且主队列中，每个子队列的工作永远只有一个，这是算法的必然结果。

3. 源码简析

咱们用最简略的 WFGoTask 为例子，把形象的调度算法从外到里一层层落实到代码上。

1) 用法示例

void func(int a);

// 应用时
WTGoTask *task = WFTaskFactory::create_go_task("queue_name_test", func, 4); 
task->start();

2) 派生关系

理解过 Workflow 工作的小伙伴肯定晓得，Workflow 任何工作都是 行为派生 ，而两头有一层，是 根本单元 ，即由SubTask 和具体 执行单元 双派生，这样既能够让下层工作被 SubTask 串到工作流里、也能够做具体执行单元做的事件。

对计算调度来说，具体 执行单元 那必定是每个能够被线程调度起来的计算工作。

咱们能够看到 WFGoTask 是从 ExecRequest 派生的，而 ExecRequest 就是执行的根本单元。（温习到 网络层面 ，根本单元是CommRequest，一个代表 执行 ，一个代表 网络 ，对称性无处不在～）
关上 src/kernel/ExecRequest.h 文件能够找到它，这里只看 dispatch() 里做了什么：

 // 这里能够看到，具体执行单元是 ExecSession，它负责和 Executor 等打交道
class ExecRequest : public SubTask, public ExecSession 
{
public:
    virtual void dispatch()
    {if (this->executor->request(this, this->queue) < 0)
        ...
    }

dispath()做的事件，就是把本人和本人的队列，通过 request() 接口提交到 Executor。

3) Executor 的生产接口：request()

int Executor::request(ExecSession *session, ExecQueue *queue)
{
    ...
    // 把工作加到对应的子队列前面  
    list_add_tail(&entry->list, &queue->task_list);

    if (queue->task_list.next == &entry->list)
    { // 子队列方才是空的，那么能够立即把此工作提交到 Executor 中
        struct thrdpool_task task = {
        .routine = Executor::executor_thread_routine,
        .context = queue
        };

        if (thrdpool_schedule(&task, this->thrdpool) < 0)  // 提交
            ...
    }  
    return -!entry;  
}

4) Executor 的生产接口：routine()

方才看到线程真正执行一个工作的时候，是调用的executor_thread_routine()，传进去的 context 就是这个工作所在的子队列。

void Executor::executor_thread_routine(void *context)
{ExecQueue *queue = (ExecQueue *)context;
    ...
    entry = list_entry(queue->task_list.next, struct ExecTaskEntry, list);  
    list_del(&entry->list); // 从子队列里删掉以后正要执行的工作
    ... 
    if (!list_empty(&queue->task_list))
    { // 如果子队列前面还有工作（也就是同名工作），放进来主队列中
        struct thrdpool_task task = {
        .routine = Executor::executor_thread_routine,
        .context = queue
        }; 
        __thrdpool_schedule(&task, entry, entry->thrdpool);
    }
    ...
    session->execute(); // 执行当前任务... 跑啊跑...
    session->handle(ES_STATE_FINISHED, 0); // 执行完，调用接口，会买通后续工作流
}

4. 革新案例分享：用工作流代替传统流水线模式

在公司外部，最经典的革新案例就是用 Workflow 的任务调度替换掉传统的流水线模式，和结尾介绍的按资源比例调配不同模块的线程数量是相似的，这对于某一个步骤突发减少的耗时、想要额定减少另一个步骤 /module 等，都是十分不灵便的计划。

而 Workflow 的计划相比起来，则能够完满避开传统做法的许多弊病。

除此之外，理论的计算调度中还有一些问题，是十分考验框架的实现细节的。比方，错误处理好不好做，依赖和勾销好不好做，生命周期好不好治理。尽管这些不是计算模块自身的事件，但 Workflow 的工作流这层都提供了很好的解决方案。

在上一篇线程池的文章 po 上网时，也有小伙伴问到：

做法如下简略分享一下：

5. 最初

如上总结一些心得吧。

咱们做计算调度时往往忘了，基本 要解决问题是 A ->B->C，而不是 关怀 1:2:3。如果经常要放心其余问题，往往是因为调度计划自身做得不够通用。只有一个最通用、最回归问题实质的架构计划，才能够让开发者不必关怀其余问题，专一于晋升本人的模块自身，也更不便下层做二次开发，为开发者提供充斥想象力的无穷可能性。

另外也是作为上一篇逻辑齐备的线程池的场景补充吧，把 Executor 等下层代码拿进去剖析，能力真正感触到底层线程池中让工作自身能够调起下一个工作等做法的重要性。

Workflow 外部有许多翻新的做法，兴许我自身有许多表白不好或者技术不到位的中央，但技术文章都是抱着一种分享新思路新做法的心态~ 那么，很心愿能够看到大家的不同意见，欢送发到我的项目的 issue 中~~~

https://github.com/sogou/workflow