[Feature][Master+API+Scheduler] Propose for master refactor and scheduler module #4355
Comments
|
the whole structure:
|
lenboo
changed the title
|
关于Scheduler模块 我们考虑另外一种方案： hash分配cmd，去掉scheduler
|
About the Scheduler module We consider another option:
|
|
Hi:
|
1 similar comment
|
Hi:
|
|
+1 Sounds good to me. |
|
It seems like a huge challenge, but it's worth focusing on |
yup, do you have any ideas about this (design or involvement in the development of the code)? |
|
@CalvinKirs sorry for the delay, If there is a chance, I am willing to participate the development. |
|
this work already have been done. so i close this issue. |
Backgroud:
refer: #4083
At present, the problems of master:
There are many polling, that result in unnecessary time-cost
The distributed lock is used when the command is taken, that result in the bottleneck of concurrency
Too many threads(nProcessInstance*nTaskInstances) are used, that result in the waste of system resources
Polling database result in database query pressure bottleneck
here are the propose:
1. Reconstruct the communication function
Synchronization: sending thread sends message (blocking) - > the receiver receives the message and processes the message (storing dB, writing data, etc.) - > the receiver returns the message to the sender - > the sender unblocks
Asynchronous: send thread send message, send thread cache message, receive message and process message, reply to sender command after processing message, remove cache message after receiving command
2. Add scheduler function
Realize ha function
To implement the scheduler start process, scan the CMD table first, and then start the monitoring function of CMD to cache the CMD data to the local queue
Monitor the receiving process of CMD (synchronization)
CMD caches the queue processing thread and sends the CMD to the master according to the policy
The implementation of CMD sending policy can support multiple policies at the same time, and can be easily extended, such as the priority of CMD, the load of master, and so on
3. API
4. Fault tolerant process modification:
Master fault tolerance
Workflow instance responsible for fault tolerant master: find the unfinished workflow instance, generate fault tolerant CMD, and send it to active scheduler
Find the unfinished workflow instance and generate the fault-tolerant CMD
Send CMD to active scheduler
Find the unfinished command and send it to the scheduler for reprocessing (remove the host and reassign the master)
The unfinished command is found according to the host of the dead Master
Send the command back to the scheduler and let the scheduler reallocate it
Worker fault tolerance
When the worker hangs up, the master listens to the message and finds out all the unfinished tasks on the worker from the DB
Judge whether the DAG of the task belongs to itself, and change the state of the task to "fault tolerant"
Trigger task status modification
5. Master execution process
Modify the process pool of master processing workflow, from obtaining CMD to generating workflow instance to starting task submission to the end of queue
When the master submits a task, it is found that the task is in progress and needs to be updated
Check (whether the task instance worker is alive + worker start time < task start time). If the worker has fault tolerance, modify the task state to "fault tolerance" and trigger the task state change
Inform the worker where the task is, and change the host reported by the worker to the current master
Add master task status monitoring
Receive task / workflow status from worker / API / Master
Save the received task status to the corresponding DAG's unprocessed status queue
The corresponding DAG cannot be found in the current task state, and the receive failure is required
Add the thread pool of master task state processing, all task states of the same workflow can only be processed sequentially
When the master receives a task, it needs to determine whether the DAG to which the task belongs is processed by the master. If not, it does not belong to its own processing and needs to discard this status
Determine whether a thread is already processing the DAG state. If it already exists, the DAG state will not be processed
Determine whether the task state exists in the list of unprocessed states in DAG. If not, discard the state
Start the thread to process the DAG task status, and add DAG as key and feature as value to the map to facilitate the previous judgment
After processing the task state, delete the map and release the thread
The polling thread is added to poll the external workflow / task status (dependent / sub workflow).
Query the status of dependent tasks
Query whether the dependent task component is successful
According to the combination of dependency conditions, the dependency state is generated
Query subworkflow status
Query parent child relationship table
Query the status of a subworkflow
Modify the state of a subtask based on the state of the subworkflow
Add a timeout monitoring queue, add the task / workflow that needs to be monitored to the queue, and a time wheel / thread will monitor the timeout. If a timeout occurs, the timeout processing will be triggered.
The newly submitted task adds a timeout monitoring task in the timeout queue
If the task is in the timeout queue and the timeout is triggered, it will alarm or kill the task according to the timeout policy
When the task status is completed, it is removed from the timeout queue
The master monitors the CMD thread, marks the CMD, and caches the CMD to the local CMD queue.
The CMDS that cannot be processed by the master can be fed back to the scheduler for redistribution
Actively report the resource usage, and the master reports the resource usage to the scheduler.
6. Timing
Master timing: to prevent a timing from triggering multiple times: there may be multiple times of timing when the master sends the timing to the scheduler.
Using quartz distributed timing, each trigger timing
Add a unique index (definitionid + schedulertime + datetime) to the CMD table and the workflow instance table to prevent duplication.
==================================
1. 重构通信功能
同步： 发送线程send消息（阻塞） -> 接收方收到消息，并且处理消息（存db,写数据等）->接收方返回消息给发送方 -> 发送方解除阻塞
异步： 发送线程send消息->发送线程缓存消息-> 接收方收到消息，并处理->处理完消息回复发送方command -> 发送方收到command，移除缓存消息
2. 新增scheduler功能
实现ha功能
实现scheduler启动流程,先扫描cmd表，再启动监听cmd功能，将cmd数据缓存到本地队列
监听cmd接收处理流程（同步）
cmd缓存队列处理线程，将cmd根据策略发送给master
实现cmd发送策略，可以同时支持多个策略，且可以易扩展策略，例如:cmd的优先级，master的负载等等
3.API部分
4. 容错流程修改：
master容错
容错master负责的工作流实例： 找到未完成的工作流实例，生成容错cmd,将cmd发给active的scheduler.
找到还未处理完的command,发送到scheduler进行重新处理（去掉host,重新分配master）
worker容错
5. master执行流程
修改master处理工作流线程池，从获取cmd到生成工作流实例，到开始任务全部提交到队列结束
增加master任务状态监听
增加master任务状态处理线程池，同一个工作流的所有任务状态只能顺序处理
增加轮询线程，针对需要获取外部工作流/任务状态的需求（依赖/子工作流），进行轮询。
增加超时监控队列，将需要监控的任务/工作流加入队列，由一个时间轮/线程进行超时监控，发生超时，则触发超时处理。
master监听cmd线程，对cmd标记，cmd缓存到本地cmd队列。
master处理不完的cmd可以反馈回scheduler进行重新分发
主动上报资源使用情况，master向scheduler汇报资源使用情况。
6. 定时
The text was updated successfully, but these errors were encountered: