webrtc的定时任务RepeatingTaskHandle实现过程

大家好，今天看下webrtc定时任务的实现。

RepeatingTaskHandle 定时任务

首先看下类的声明，具体文件见：repeating_task.h。

RepeatingTaskHandle 提供了两个函数实现定时任务

Start(*) 函数
DelayedStart(*) 函数

这两个函数实现原理一样，唯一的区别是DelayedStart()会延后一个指定时间，再开启定时任务

// Allows starting tasks that repeat themselves on a TaskQueue indefinately
// until they are stopped or the TaskQueue is destroyed. It allows starting and
// stopping multiple times, but you must stop one task before starting another
// and it can only be stopped when in the running state. The public interface is
// not thread safe.
class RepeatingTaskHandle {
 public:
  RepeatingTaskHandle() = default;
  ~RepeatingTaskHandle() = default;
  RepeatingTaskHandle(RepeatingTaskHandle&& other);
  RepeatingTaskHandle& operator=(RepeatingTaskHandle&& other);
  RepeatingTaskHandle(const RepeatingTaskHandle&) = delete;
  RepeatingTaskHandle& operator=(const RepeatingTaskHandle&) = delete;

  // Start can be used to start a task that will be reposted with a delay
  // determined by the return value of the provided closure. The actual task is
  // owned by the TaskQueue and will live until it has been stopped or the
  // TaskQueue is destroyed. Note that this means that trying to stop the
  // repeating task after the TaskQueue is destroyed is an error. However, it's
  // perfectly fine to destroy the handle while the task is running, since the
  // repeated task is owned by the TaskQueue.
  
  //Start()函数，开启一个定时任务
  //注意该函数是个模板函数，所以可以开启任意的定时任务
  //内部其实是创建了一个RepeatingTaskImpl的 unique_ptr智能指针
  //然后把该智能指针抛到上一篇介绍的 task_queue 队列中
  //task_queue对应的线程会执行该repeating_task任务
  //所有的定时，其实是有RepeatingTaskImpl类实现的
  //大家会发现，这个Start()函数，并没有传定时器的时间间隔
  //其实这个时间间隔，是需要你传定时函数Closure<>时，在内部指定并返回
  //后面会具体分析
  template <class Closure>
  static RepeatingTaskHandle Start(TaskQueueBase* task_queue,
                                   Closure&& closure,
                                   Clock* clock = Clock::GetRealTimeClock()) {
    auto repeating_task = std::make_unique<
        webrtc_repeating_task_impl::RepeatingTaskImpl<Closure>>(
        task_queue, TimeDelta::Zero(), std::forward<Closure>(closure), clock);
    auto* repeating_task_ptr = repeating_task.get();
    task_queue->PostTask(std::move(repeating_task));
    return RepeatingTaskHandle(repeating_task_ptr);
  }

  // DelayedStart is equivalent to Start except that the first invocation of the
  // closure will be delayed by the given amount.
  //DelayedStart()和前面的Start()函数几乎一样
  //只是添加了一个延后执行时间 first_delay
  //然后调用PostDelayedTask抛给task_queue的延时队列，   
  //延时一个first_delay时间，然后再执行定时任务
  template <class Closure>
  static RepeatingTaskHandle DelayedStart(
      TaskQueueBase* task_queue,
      TimeDelta first_delay,
      Closure&& closure,
      Clock* clock = Clock::GetRealTimeClock()) {
    auto repeating_task = std::make_unique<
        webrtc_repeating_task_impl::RepeatingTaskImpl<Closure>>(
        task_queue, first_delay, std::forward<Closure>(closure), clock);
    auto* repeating_task_ptr = repeating_task.get();
    task_queue->PostDelayedTask(std::move(repeating_task), first_delay.ms());
    return RepeatingTaskHandle(repeating_task_ptr);
  }
  ....
};

RepeatingTaskImpl

向任务队列task_queue中抛入的正是RepeatingTaskImpl，可以看到该类继承自 RepeatingTaskBase，而该基类重写了任务执行的Run()函数，来真正的实现定时任务。


// The template closure pattern is based on rtc::ClosureTask.
template <class Closure>
class RepeatingTaskImpl final : public RepeatingTaskBase {
 //RunClosuer()函数中，执行前文提到的的定时任务函数 closure_
 private:
  TimeDelta RunClosure() override { return closure_(); }

  typename std::remove_const<
      typename std::remove_reference<Closure>::type>::type closure_;
};

基类 RepeatingTaskBase 的声明：

class RepeatingTaskBase : public QueuedTask {
 public:
  RepeatingTaskBase(TaskQueueBase* task_queue,
                    TimeDelta first_delay,
                    Clock* clock);
  ~RepeatingTaskBase() override;

  void Stop();

 private:
  virtual TimeDelta RunClosure() = 0;
  //task_queue线程发现有任务时，会调用该任务的Run()函数
  bool Run() final;

  TaskQueueBase* const task_queue_;
  Clock* const clock_;
  // This is always finite, except for the special case where it's PlusInfinity
  // to signal that the task should stop.
  Timestamp next_run_time_ RTC_GUARDED_BY(task_queue_);
};

定时任务实现的核心是Run()函数。

其实原理也很简单：所谓定时任务就是执行完该任务后，再重新把该任务抛给 task_queue_，这样任务就会不停地被执行了。

bool RepeatingTaskBase::Run() {
  RTC_DCHECK_RUN_ON(task_queue_);
  // Return true to tell the TaskQueue to destruct this object.
  if (next_run_time_.IsPlusInfinity())
    return true;
  //首先会执行用户抛进来的定时任务函数
  //有一个特别重要的点，定时任务执行完后会返回一个delay时间
  //这个时间就是定时器执行的间隔时间
  TimeDelta delay = RunClosure();

  // The closure might have stopped this task, in which case we return true to
  // destruct this object.
  //如果该任务被Stop()了，就返回不再执行
  if (next_run_time_.IsPlusInfinity())
    return true;

  RTC_DCHECK(delay.IsFinite());
  //定时任务执行所用时间 = 当前时间 减去 定时任务开始执行时间
  TimeDelta lost_time = clock_->CurrentTime() - next_run_time_;
  //定时任务下次执行时间 = 上次执行时间 加上 定时器的间隔时间
  next_run_time_ += delay;
  //但是因为定时任务本身执行也耗时，所以真正执行的时间要减去定时任务的执行时间
  delay -= lost_time;
  delay = std::max(delay, TimeDelta::Zero());
  //然后再重新把该任务抛入到延时队列中
  //这样定时任务就能被一直执行
  task_queue_->PostDelayedTask(absl::WrapUnique(this), delay.ms());

  // Return false to tell the TaskQueue to not destruct this object since we
  // have taken ownership with absl::WrapUnique.
  return false;
}

void RepeatingTaskBase::Stop() {
  RTC_DCHECK_RUN_ON(task_queue_);
  RTC_DCHECK(next_run_time_.IsFinite());
  next_run_time_ = Timestamp::PlusInfinity();
}

下面看一个例子

gcc 码率预估模块，就是开启了一个定时器不停去更新最新码率

具体见：rtp_transport_controller_send.cc

首先声明一个任务队列 rtc::TaskQueue task_queue_;
和定时任务管理类RepeatingTaskHandle controller_task_
然后向该任务队列中抛入定时任务

controller_task_ = RepeatingTaskHandle::DelayedStart(
        task_queue_.Get(), process_interval_, [this]() {
          RTC_DCHECK_RUN_ON(&task_queue_);
          UpdateControllerWithTimeInterval();
          return process_interval_;
        });

定时函数返回的process_interval_就是定时器的时间间隔。

以上就是webrtc的定时任务RepeatingTaskHandle 实现过程了。

欢迎指正，欢迎关注，感谢大家！