图框架task任务调度状态机 TaskStateenumclassTaskState{Start,// 未调度开始状态Scheduled,// 已经被调度器调度尚未开始执行Waiting,// 等待态Ready,// 就绪态尚未调度Running,// 执行中Fallback,// 兜底态Finishing,// 运行结束正在生成结果Ok,// 下面是具体的结果状态Fail,Timeout,Cancel};状态机同步任务同步任务状态切换Start - Scheduled - Running - Finishing - Ok/Fail/Timeout/Cancel异步任务Start - Scheduled - Running - Waiting - [Resume] - Ready - Scheduled - …如果在Running和Waiting之间调用了Resume则它会跳过Waiting状态直接发起调度Start - Scheduled - Running - [Resume] - Ready - [由之前正在Running的线程重新发起调度] - Scheduled -异步算子设计异步算子的exec函数会执行2次。第一次会调用brpc异步函数执行远程rpc调用并将算子状态改成Waiting状态。当远程rpc回调函数返回调用schedule的ResumeTask函数将算子状态改成TaskState::Ready调度器调度改成TaskState::Scheduled状态第二次执行该算子的exec函数此时将算子返回的表输出。StatusProxyRemoteOp::Exec(vectorTableinputs,vectorTableoutput,conststd::shared_ptrGraphSessiongraph_session){std::unique_ptrdoe::brpc::SelectRequestproxy_remote_request(newdoe::brpc::SelectRequest);if(status_.IsPrepare()){status_Status::Waiting();autocallback[this,graph_session](Status st,vectorTabletable){std::lock_guardstd::mutex_(mutex_);output_std::move(table);status_st;graph_session-schedule_.ResumeTask(graph_session,this);};remote_op_client_-Search(GetTaskName(),std::move(proxy_remote_request),graph_session,callback);}std::lock_guardstd::mutex_(mutex_);if(!status_.IsWaiting()){outputstd::move(output_);}returnstatus_;}GraphSchedulerun调度session设置超时执行函数执行入读为0的算子GraphSchedule::Run(conststd::shared_ptrGraphSessionsession){std::weak_ptrGraphSessionweak_sessionstd::weak_ptrGraphSession(session);for(constautonode:session-runtime_graph_.GetAllNodes()){autotasknode.get();if(task-deadline_ms_0){Scheduler::GetInstance().Schedule(std::chrono::milliseconds(task-deadline_ms_),[weak_session,task](){autosessionweak_session.lock();if(sessiontask-state_TaskState::Finishing){session-schedule_.TaskFinish(nullptr,session,task,{},Status::Error(%s deadline,task-GetTaskName().c_str()),TaskState::Timeout);}});}if(task-GetInDegree()0){Schedule(nullptr,session,task);}}}Schedule函数节点第一次调度需要设置超时如果当前线程可调度直接调度到当前线程 否则线程池调度voidGraphSchedule::Schedule(GraphFrame*frame,conststd::shared_ptrGraphSessionsession,Task*node){while(!ok){TaskState statenode-state_;switch(state){caseTaskState::Start:{oknode-state_.compare_exchange_strong(state,TaskState::Scheduled);}caseTaskState::Ready:{// 异步算子Ready的情况oknode-state_.compare_exchange_strong(state,TaskState::Scheduled);break;}default:return;}}if(frameframe-next_task_nullptrframe-session_session){frame-next_task_node;}else{bthread_t bid;// 调用bthread执行函数到线程池运行bthread_start_background(bid,NULL,[](void*args)-void*{std::unique_ptrLocalVarsvars((LocalVars*)args);GraphFrameframe(vars-session);frame.next_task_vars-node;frame.Run();returnNULL;},newLocalVars{node,session});}}ResumeTask用于异步算子的调用将Running和Waiting状态设为Ready状态voidGraphSchedule::ResumeTask(conststd::shared_ptrGraphSessionsession,Task*task){boolokfalse;while(!ok){TaskState statetask-state_;switch(state){caseTaskState::Running:{oktask-state_.compare_exchange_strong(state,TaskState::Ready);// running结束后会自动重新调度break;}caseTaskState::Waiting:{oktask-state_.compare_exchange_strong(state,TaskState::Ready);if(ok){Schedule(nullptr,session,task);}break;}default:return;}}}TaskFinish该算子执行完成则调度下面的算子。voidGraphSchedule::TaskFinish(GraphFrame*frame,conststd::shared_ptrGraphSessionsession,Task*task,vectorTableoutput,Status status,TaskState new_state){TaskState prev_statetask-state_;TaskState next_statenew_stateTaskState::Ok||!FallbackEnabled()?TaskState::Finishing:TaskState::Fallback;do{}while(!task-state_.compare_exchange_strong(prev_state,next_state));// 如果成功或者允许失败的话开始调度后继节点if(task-state_TaskState::Ok||task-allow_fail){for(autonext:task-GetOutNodes()){if(next-front_finish_num_next-GetInDegree()){Schedule(frame,session,next);}}}else{// 整个session失败掉session-Finish(status,task-output_result_);}}GraphFrame用于bthread调度执行图任务算子classGraphFrameGraphFrame(std::shared_ptrGraphSessionsession):session_(session)voidRun();std::shared_ptrGraphSessionsession_;Task*cur_task_nullptr,*next_task_nullptr;}voidGraphFrame::Run(){while(next_task_!nullptr!session_-finished_){// run之前这个任务一定是通过ScheduleTask进入Scheduled状态autoscheduledTaskState::Scheduled;if(!cur_task_-state_.compare_exchange_strong(scheduled,TaskState::Running)){return;}statuscur_task_-DoExec(inputs,output,session_);if(status.IsWaiting()){TaskState runningTaskState::Running;if(!cur_task_-state_.compare_exchange_strong(running,TaskState::Waiting){// 失败的话意味着在DoExec之前就已经进入了Ready态此时需要直接调度当然也有可能已经Finish了Schedule会处理这些其他情况session_-schedule_.Schedule(this,session_,cur_task_);}}else{session_-schedule_.TaskFinish(this,session_,cur_task_,std::move(output),status,status.IsOk()?TaskState::Ok:TaskState::Fail);}}}
