apollo介绍之Cyber定时器(四)

TimerOption(uint32_tperiod,std::function<void()>callback,booloneshot):period(period),callback(callback),oneshot(oneshot){}

voidTimer::Start(){// 1. 首先判断定时器是否已经启动
if(!started_.exchange(true)){// 2. 初始化任务
if(InitTimerTask()){// 3. 在时间轮中增加任务
timing_wheel_->AddTask(task_);AINFO<<"start timer ["<<task_->timer_id_<<"]";}}}

boolTimer::InitTimerTask(){// 1. 初始化定时任务
task_.reset(newTimerTask(timer_id_));task_->interval_ms=timer_opt_.period;task_->next_fire_duration_ms=task_->interval_ms;// 2. 是否单次触发
if(timer_opt_.oneshot){std::weak_ptr<TimerTask>task_weak_ptr=task_;// 3. 注册任务回调
task_->callback=[callback=this->timer_opt_.callback,task_weak_ptr](){autotask=task_weak_ptr.lock();if(task){std::lock_guard<std::mutex>lg(task->mutex);callback();}};}else{std::weak_ptr<TimerTask>task_weak_ptr=task_;// 注册任务回调
task_->callback=[callback=this->timer_opt_.callback,task_weak_ptr](){std::lock_guard<std::mutex>lg(task->mutex);autostart=Time::MonoTime().ToNanosecond();callback();autoend=Time::MonoTime().ToNanosecond();uint64_texecute_time_ns=end-start;if(task->last_execute_time_ns==0){task->last_execute_time_ns=start;}else{// start - task->last_execute_time_ns 为2次执行真实间隔时间，task->interval_ms是设定的间隔时间
// 注意误差会修复补偿，因此这里用的是累计，2次误差会抵消，保持绝对误差为0
task->accumulated_error_ns+=start-task->last_execute_time_ns-task->interval_ms*1000000;}task->last_execute_time_ns=start;// 如果执行时间大于任务周期时间，则下一个tick马上执行
if(execute_time_ms>=task->interval_ms){task->next_fire_duration_ms=TIMER_RESOLUTION_MS;}else{int64_taccumulated_error_ms=::llround(static_cast<double>(task->accumulated_error_ns)/1e6);if(static_cast<int64_t>(task->interval_ms-execute_time_ms-TIMER_RESOLUTION_MS)>=accumulated_error_ms){// 这里会补偿误差
task->next_fire_duration_ms=task->interval_ms-execute_time_ms-accumulated_error_ms;}else{task->next_fire_duration_ms=TIMER_RESOLUTION_MS;}}TimingWheel::Instance()->AddTask(task);};}returntrue;}

512个bucket64个roundtick为2ms

voidTimingWheel::AddTask(conststd::shared_ptr<TimerTask>&task,constuint64_tcurrent_work_wheel_index){// 1.不是运行状态则启动时间轮
if(!running_){// 2.启动Tick线程，并且加入scheduler调度。
Start();}// 3. 计算一下轮bucket编号
autowork_wheel_index=current_work_wheel_index+task->next_fire_duration_ms/TIMER_RESOLUTION_MS;// 4. 如果超过最大的bucket数
if(work_wheel_index>=WORK_WHEEL_SIZE){autoreal_work_wheel_index=GetWorkWheelIndex(work_wheel_index);task->remainder_interval_ms=real_work_wheel_index;autoassistant_ticks=work_wheel_index/WORK_WHEEL_SIZE;// 5.疑问，如果转了一圈之后，为什么直接加入剩余的bucket？？？
if(assistant_ticks==1&&real_work_wheel_index!=current_work_wheel_index_){work_wheel_[real_work_wheel_index].AddTask(task);ADEBUG<<"add task to work wheel. index :"<<real_work_wheel_index;}else{autoassistant_wheel_index=0;{// 6.如果超出，则放入上一级时间轮中
std::lock_guard<std::mutex>lock(current_assistant_wheel_index_mutex_);assistant_wheel_index=GetAssistantWheelIndex(current_assistant_wheel_index_+assistant_ticks);assistant_wheel_[assistant_wheel_index].AddTask(task);}ADEBUG<<"add task to assistant wheel. index : "<<assistant_wheel_index;}}else{// 7. 如果没有超过最大bucket数，则增加到对应的bucket中
work_wheel_[work_wheel_index].AddTask(task);ADEBUG<<"add task ["<<task->timer_id_<<"] to work wheel. index :"<<work_wheel_index;}}

voidTimingWheel::TickFunc(){Raterate(TIMER_RESOLUTION_MS*1000000);// ms to ns
// 1. 循环调用
while(running_){// 2. 执行bucket中的回调，并且删除当前bucket中的任务(回调中会增加新的任务到bucket)
Tick();tick_count_++;// 3. 休眠一个Tick
rate.Sleep();{std::lock_guard<std::mutex>lock(current_work_wheel_index_mutex_);// 4.获取当前bucket id，每次加1
current_work_wheel_index_=GetWorkWheelIndex(current_work_wheel_index_+1);}// 5.下一级时间轮已经转了一圈，上一级时间轮加1
if(current_work_wheel_index_==0){{// 6.上一级时间轮bucket id加1
std::lock_guard<std::mutex>lock(current_assistant_wheel_index_mutex_);current_assistant_wheel_index_=GetAssistantWheelIndex(current_assistant_wheel_index_+1);}// 7. 
Cascade(current_assistant_wheel_index_);}}}