Task Manager 的设计简述

讲解 Task Manager 之前，在这里先介绍一些 Task Manager 会使用到的概念术语。

图数据库 Nebula Graph 中，存在一些长期在后台运行的任务，我们称之为 Job 。存储层存在的 DBA 使用的部分指令，比如：数据完成导入后，想在全局做一次 compaction，都是 Job 范畴。

作为一个分布式的系统，Nebula Graph 中 Job 由不同的 storaged 完成，而我们管一个 storaged 上运行的 Job 子任务叫做 Task 。Job 的控制由 metad 上的 Job Manager 负责，而 Task 的控制由 storaged 上的 Task Manager 负责。

在本文中，我们着重讲述如何对长耗时的 Task 进行管理与调度进一步提升数据库性能。

Task Manager 要解决的问题

上文说到 storaged 上的 Task Manager 控制的 Task 是 meta 控制的 Job 的子任务，那 Task Manager 它自己具体解决什么问题呢？在 Nebula Graph 中 Task Manager 主要解决了以下 2 个问题：

将之前通过 HTTP 的传送方式改为 RPC （ Thrift ）一般用户在搭建集群时，知道 storaged 之间通信使用 Thrift 协议，会为 Thrift 所需端口开放防火墙，但是可能意识不到 Nebula Graph 还需要使用 HTTP 端口，我们遇到过多次社区用户实践忘记开放 HTTP 端口的事情。
storaged 对于 Task 有调度能力这块内容将在本文下面章节展开讲述。

Task Manager 在 Nebula Graph 中的位置

Task Manager 体系中的 meta

在 Task Manager 体系中, metad （ JobManager ）的任务是根据 graphd 中传过来的一个 Job Request，选出对应的 storaged host，并拼组出 Task Request 发给对应的 storaged 。不难发现，体系中 meta 接受 Job Request，拼组 Task Request , 发送 Task Request 及接受 Task 返回结果，这些逻辑的套路是稳定的。而如何拼组 TaskRequest，将 Task Request 发给哪些 storaged 则会根据不同的 Job 有所变化。JobManager 用 模板策略 + 简单工厂 以应对未来的扩展。

让未来的 Job 同样继承于 MetaJobExecutor，并实现 prepare() 和 execute() 方法即可。

Task Manager 的调度控制

之前提到的，Task Manager 的调度控制希望做到 2 点：

系统资源足够时，尽可能的高并发执行 Task
系统资源吃紧时，让所有运行中的 Task 占用的资源不要超过某一个设定的阈值。

高并发执行 Task

Task Manager 将系统资源中自己持有的线程称之为 Worker 。Task Manager 有一个现实中的模拟原型——银行的营业厅。想象一下, 我们去银行办业务时会有以下几步:

场景 1：在门口的排号机拿一个号
场景 2：在大厅找个位置, 边玩手机边等叫号
场景 3：等叫到号时, 到指定窗口办理

同时, 你还会碰到这样那样的问题:

场景 4：VIP 可以插队
场景 5：你可能排着队, 因为某些原因, 放弃了本次业务
场景 6：你可能排着排着队, 银行就关门了

那么, 整理一下, 这也就是 Task Manager 的基本需求

Task 按 FIFO 顺序执行：不同的 Task 有不同的优先级，高优先级的可以插队
用户可取消一个排队中的 Task
storaged 随时 shutdown
一个 Task，为了使其尽可能高的并发，会被拆分为多个 SubTask，SubTask 是每个 Worker 真正执行的任务
Task Manager 是全局唯一实例，要考虑多线程安全性

于是, 有了如下实现:

实现 1：用 Thrift 结构中的 JobId 和 TaskId，确定一个 Task，称为 Task Handle 。
实现 2：TaskManager 会有一个 Blocking Queue，负责让 Task 的 Handle 排队执行（排号机），而 Blocking Queue 本身线程安全。
实现 3：Blocking Queue 同时支持不同的优先级, 高优先级先出队（ VIP 插队的功能）。
实现 4：Task Manager 维持一个全局唯一的 Map，key 是 Task Handle，value 是具体的 Task （银行的大厅）。在 Nebula Graph 中采用了 folly 的 Concurrent Hash Map，线程安全的 Map 。
实现 5：如果有用户 cancel Task，直接在根据 Handle 找到 Map 中对应的 Task，并标记 cancel，对 queue 中的 Handle 不做处理。
实现 6：如果有正在运行的 Task，对于 storaged 的 shutdown 会等到这个 Task 正在执行的 subTask 执行完毕才返回。

限定 Task 占用的资源阈值

保证不超过阈值还是很简单的，因为 Worker 就是线程，只要让所有的 Worker 都出自一个线程池，就可以保证最大的 Worker 数。麻烦的是将子任务平均地分配到 Worker 中, 我们来讨论下方案:

方法一：使用 Round-robin 添加任务

最简单的方法是用 Round-robin 的方式来添加任务。也就是将 Task 分解为 Sub Task 之后, 依次追加到现在的各个 Worker 中。

但是可能会有问题, 比如说, 我有 3 个 Worker, 2 个 Task （蓝色为 Task 1，黄色为 Task 2 ）：

Round-robin 图 1

假如 Task 2 中的 Sub Task 执行远快于 Task1 的, 那么好的并行策略应该是这样：

Round-robin 图 2

简单粗暴的 Round-robin 会让 Task 2 的完成时间依赖于 Task 1 （见 Round-robin 图 1 ）。

方法二：一组 worker 处理一个 Task

针对方法一可能会出现的情况，设定专门的 Worker 只处理指定的 Task，从而避免多个 Task 相互依赖问题。但是依然不够好, 比如说：

很难保证每个 Sub Task 执行时间基本相同，假设 Sub Task 1 的执行明显慢于其他的 Sub Task，那么好的执行策略应该是这样的：

这个方案还是避免不了 1 核有难，10 核围观的问题 👀。

方法三：Nebula Graph 采用的解决方案

在 Nebula Graph 中 Task Manager 会将 Task 的 Handle 交给 N 个 Worker 。N 由总 Worker 数、总 Sub Task 数，以及 DBA 在提交 Job 时指定的并发参数共同决定。

每个 Task 内部维护一个 Blocking Queue （下图的 Sub Task Queue ），存放 Sub Task 。Worker 在执行时，根据自己持有的 Handle 先找到 Task，再从 Task 的 Block Queue 中获取 Sub Task 。

设计补充说明

问题 1: 为什么不直接将 Task 放到 Blocking Queue 排队，而是拆成两部分，将 Task 保存在 Map 里，让 Task Handle 排队?

主要原因是 C++ 多线程基础设施不好支持这种逻辑。Task 需要支持 cancel 。假设 Task 放在 Blocking Queue 中，就需要 Blocking Queue 支持定位到其中的某一个 Task 的能力。而当前 folly 中的 Blocking Queue 都没有此类接口。

问题 2: 什么样的 Job 有 VIP 待遇?

当前 Task Manager 支持的 compaction / rebuild index 对执行时间并不敏感，支持类似 count() 查询操作功能尚在开发中。考虑到用户希望在一个相对短的时间内完成 count() ，那么假如正好碰上了 storaged 在做多个 compaction，还是希望 count(*) 可以优先运行，而非在所有 compaction 之后再开始做。

本文中如有任何错误或疏漏欢迎去 GitHub：https://github.com/vesoft-inc/nebula issue 区向我们提 issue 或者前往官方论坛：https://discuss.nebula-graph.com.cn/ 的 建议反馈 分类下提建议 👏；加入 Nebula Graph 交流群，请联系 Nebula Graph 官方小助手微信号：NebulaGraphbot

作者有话说：Hi，我是我是 lionel.liu ，是图数据 Nebula Graph 研发工程师，对数据库查询引擎有浓厚的兴趣，希望本次的经验分享能给大家带来帮助，如有不当之处也希望能帮忙纠正，谢谢~