Trino中Task源码解析(tars源码分析)
我们知道,在Trino中一个Query会拆分成多个Stage,一个Stage又会拆分成多个Task,Task是跑在Worker上的具体任务,那一个Task周围有哪些息息相关的类和方法呢,需要我们去阅读源码分析。
整体框架
和Task关系密切的几个类以及关系如下图所示
TaskResource
Task的创建,删除,更新都是通过Http请求来完成,由TaskResource这个类来接受请求, 但具体的实现方法都封装在SqlTaskManager中,TaskResource接收到请求后调用SqlTaskManager中对应的方法,以创建Task为例:
SqlTaskManager
该类中有许多对Task进行操作的方法,比如创建,更新,取消,中止等等
上图中的类属性tasks就是用来保存所有task相关信息的,本质是一个不可驱逐的缓存,缓存中key是taskId(每个task的专属标识), value是对应创建的sqlTask对象。
因此SqlTaskManager中对Task操作就是从tasks中根据taskId拿到SqlTask对象,再调用对应的方法。
StucksplitTasksInterrupter
SqlTaskManager中还有个很重要的特性就是StuckSplitTasksInterrupter,他会定时的去检查是否有task卡住,如果卡住则被标记为stuck并会被kill,至于是否开启这个功能以及多长时间没响应才算卡住,都可以通过参数去配置。
可以看到代码中会去遍历当前所有的runningSplit,如果该runningSplit执行时间大于设置的阈值,则会被筛选出来拿到对应的TaskId, 再调用sqlTask的fail方法,结束这个卡住的Task
SqlTask
每一个Task都对应于一个SqlTask对象,其中比较重要的几个属性:
- TaskStateMachine: 用来记录Task的状态,在调用sqlTask的cancel, abort等接口时,其实就是修改状态机的状态,并且状态机会有一个Listener监听状态机的变化,一旦有更新,就会执行相应的操作
- SqlTaskExecution: 在创建SqlTask对象时,同时也会创建一个对应的SqlTaskExecution。 SqlTaskExecution主要是负责Split到Driver的调度,并把split和TaskExecutor关联起来, 通过创建一个TaskHandle放到TaskExecutor的队列中去等待执行。在创建TaskHandle时,会添加一个Listener,如果这个Task的状态被设置为Terminating或者Done时,就会调用TaskExecutor中的removeTask方法来真正的取消这个task任务
TaskExecutor
TaskExecutor是具体执行Task任务的地方。他有一个线程池,线程池的大小可以通过参数配置,初始化的时候就会创建出对应个数的线程,每个线程执行一个TaskRunner,在TaskRunner中,while循环从waitingSplit中拿到Split,放到runningSplit中并执行这个split。前面提到的StuckSplitTasksInterrupter中获取的runningSplit信息就是从这个地方更新的。而waitingSplit中的split就是在上面提到的SqlTaskExecution中放进去的。