exec_plan_fragment
fe:
1.fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
public void exec() throws Exception {
?service FrontendService {
718 ? ? TGetDbsResult getDbNames(1:TGetDbsParams params)
719 ? ? TGetTablesResult getTableNames(1:TGetTablesParams params)
720 ? ? TDescribeTableResult describeTable(1:TDescribeTableParams params)
721 ? ? TShowVariableResult showVariables(1:TShowVariableRequest params)
? ? ? ? ? ?// be在执行plan_fragement_exexutor.cpp执行计划时候，上报执行状态
722 ? ? TReportExecStatusResult reportExecStatus(1:TReportExecStatusParams params)
723? ? ?// be在完成比如create table语句后，发送
724 ? ? MasterService.TMasterResult finishTask(1:MasterService.TFinishTaskRequest request)
? ? ? ? ? ?//
725 ? ? MasterService.TMasterResult report(1:MasterService.TReportRequest request)
be:
1.exec_plan_fragment rpc接口,_exec_env->fragment_mgr()->exec_plan_fragment( backend_service.cpp
2.runtime/fragment_mgr.cpp执行exec_plan_fragment
?std::shared_ptr<FragmentExecState> exec_state;
exec_state->prepare(params));
FragmentExecState::prepare(
_executor.prepare(params);
PlanFragmentExecutor _executor;
3.runtime/plan_fragment_executor.cpp
ExecNode::create_tree(
finish_task
fe: 将任务从AgentTaskQueue中移除
文件名master/MasterImpl.java
public TMasterResult finishTask(TFinishTaskRequest request) {
finishCreateReplica
AgentTaskQueue.removeTask(? ? 将任务从?AgentTaskQueue移除
be:完成任务发送finishTask rpc请求
文件名be/src/agent/task_worker_pool.cpp?
void TaskWorkerPool::_finish_task(const TFinishTaskRequest& finish_task_request) {? ? ?
submit_tasks
service BackendService {
AgentService.TAgentResult submit_tasks(1:list<AgentService.TAgentTaskRequest> tasks);
be:? 将task放入threadpool中出来
be/src/agent/agent_server.cpp? task_worker_pool.cpp? ? ?
void TaskWorkerPool::submit_task(const TAgentTaskRequest& task) {? ?
if (_register_task_info(task_type, signature)) { 添加当前be正在处理的任务
_finish_task(finish_task_request); 发送finishTask rpc给fe
_remove_task_info(agent_task_req.task_type, agent_task_req.signature); 将任务从be中移除
fe:
report
be:每隔interval时间上报be当前正在处理的任务
be/src/agent/agent_server.cpp? ?task_worker_pool.cpp
void TaskWorkerPool::_report_task_worker_thread_callback() {? ?
_handle_report(request, ReportType::TASK); ? ? ? ?
} while (!_stop_background_threads_latch.wait_for(
? ? ? ? ? ? MonoDelta::FromSeconds(config::report_task_interval_seconds)));? ?
fe：
rpc 接口：src/main/java/org/apache/doris/master/MasterImpl.java:? ?reportHandler.handleReport(request);
ReportHandler.java: ? ?public TMasterResult handleReport(TReportRequest request) throws TException {
?ReportTask reportTask = new ReportTask(beId, tasks, disks, tablets, reportVersion);
? ? ? ? try {
? ? ? ? ? ? ?增加任务
? ? ? ? ? ? putToQueue(reportTask);
取出任务：
? ?@Override
? ? protected void runOneCycle() {
? ? ? ? while (true) {
? ? ? ? ? ? ReportTask task = null;
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? task = reportQueue.take();
? ? ? ? ? ? ? ? task.exec();
? ? ? ? ? ? } catch (InterruptedException e) {
? ? ? ? ? ? ? ? LOG.warn("got interupted exception when executing report", e);
? ? ? ? ? ? }
? ? ? ? }
执行任务
protected void exec() {
? ? ? ? ? ? if (tasks != null) {
? ? ? ? ? ? ? ? ReportHandler.taskReport(beId, tasks);
? ? ? ? ? ? }
List<AgentTask> diffTasks = AgentTaskQueue.getDiffTasks(backendId, runningTasks);
diffTask是fe中存在的任务，而be节点中不存在的任务
?if (task.shouldResend(taskReportTime)) { // 重做任务
? ? ? ? ? ? ? ? batchTask.addTask(task);
? ? ? ? ? ? }
reportExecStatus:
runtime/fragment_mgr.cpp 上报执行心跳
void FragmentExecState::coordinator_callback(const Status& status, RuntimeProfile* profile,
bool done) {中上报
coord->reportExecStatus(res, params);
初始化FragmentExecState对象会设置callback
FragmentExecState::FragmentExecState(const TUniqueId& query_id,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?const TUniqueId& fragment_instance_id, int backend_num,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?ExecEnv* exec_env, const TNetworkAddress& coord_addr)
? ? ? ? : _query_id(query_id),
? ? ? ? ? _fragment_instance_id(fragment_instance_id),
? ? ? ? ? _backend_num(backend_num),
? ? ? ? ? _exec_env(exec_env),
? ? ? ? ? _coord_addr(coord_addr),
? ? ? ? ? _executor(exec_env, std::bind<void>(std::mem_fn(&FragmentExecState::coordinator_callback),
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? this, std::placeholders::_1, std::placeholders::_2,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? std::placeholders::_3)),
plan_fragment_executor执行的时候会上报???????
void PlanFragmentExecutor::report_profile() { 上报exec正在执行心跳
send_report(false);
