• chromium线程模型(2)-线程池实现

    通过chromium 官方文档，线程和任务一节我们可以知道 ，chromium有两类线程，一类是普通线程，最典型的就是io线程和ui线程。 另一类是 线程池线程。 今天我们先分析线程池的实现（基于版本 117.0.5847.0（开发者内部版本） （64 位） 分析）。

    通过官方文档我们知道，线程池可以用于发布串行任务，也可以发布并行任务。 和普通线程类似，线程池的模型也是从队列里面获取任务执行。 线程池的主要职责就是管理线程数量、线程生命周期、和任务分配。在chromium里面也不例外。

    chromium把任务优先级分为3级， 分别是

    • BEST_EFFORT :一般把那些后台执行不会影响用户体验的任务设置为BEST_EFFORT优先级， 比如写磁盘。
    • USER_VISIBLE: 一般把用户可见，但是不影响体验的任务设置为USER_VISIBLE，比如下载进度。
    • USER_BLOCKING：一般把影响用于体验的任务设置为USER_BLOCKING优先级，比如用户点击链接后加载并呈现网页。

    chromium把任务分为两类， 不阻塞任务和阻塞任务，不阻塞任务的特点是占用cpu但是能很快结束。 阻塞任务在执行的时候可能hold住线程，但是又不怎么占用cpu，却消耗线程数量，所以对于阻塞任务，应该临时增加线程池维护线程数量，防止不阻塞任务得不到执行。 chromium又把阻塞任务分为MAY_BLOCK（可能阻塞)任务和 WILL_BLOCK(必然阻塞任务)任务, WILL_BLOCK（可能阻塞）任务当被探测到真正阻塞的时候才会增加线程池维护线程数量。

    当线程数量不能满足需求的时候，chromium线程池会增加线程，当空闲线程空闲时间太久后就会释放空闲线程，但是还要尽量维持一些空闲线程，防止真正有任务到来的时候要现创建线程。
    另外也不能无限的创建线程，线程数量还需要有一个上限，当线程数量达到上限后，要有机制让低优先级任务主动放弃线程。

    我们下面就来分析chromium线程池是如何完成这些工作的。

    先给出数据结构。

    在这里插入图片描述

    线程池创建和线程的创建

    1、 线程的创建
    我们先来分析线程池的创建
    base/task/thread_pool/thread_pool_impl.cc

    ThreadPoolImpl::ThreadPoolImpl(StringPiece histogram_label,
                               std::unique_ptr<TaskTrackerImpl> task_tracker,
                               bool use_background_threads)
    : histogram_label_(histogram_label),
      task_tracker_(std::move(task_tracker)),
      single_thread_task_runner_manager_(task_tracker_->GetTrackedRef(),
                                         &delayed_task_manager_),
      has_disable_best_effort_switch_(HasDisableBestEffortTasksSwitch()),
      tracked_ref_factory_(this) {
  foreground_thread_group_ = std::make_unique<ThreadGroupImpl>(
      histogram_label.empty()
          ? std::string()
          : JoinString(
                {histogram_label, kForegroundPoolEnvironmentParams.name_suffix},
                "."),
      kForegroundPoolEnvironmentParams.name_suffix,
      kForegroundPoolEnvironmentParams.thread_type_hint,
      task_tracker_->GetTrackedRef(), tracked_ref_factory_.GetTrackedRef());

  if (CanUseBackgroundThreadTypeForWorkerThread()) {
    background_thread_group_ = std::make_unique<ThreadGroupImpl>(
        histogram_label.empty()
            ? std::string()
            : JoinString({histogram_label,
                          kBackgroundPoolEnvironmentParams.name_suffix},
                         "."),
        kBackgroundPoolEnvironmentParams.name_suffix,
        use_background_threads
            ? kBackgroundPoolEnvironmentParams.thread_type_hint
            : kForegroundPoolEnvironmentParams.thread_type_hint,
        task_tracker_->GetTrackedRef(), tracked_ref_factory_.GetTrackedRef());
  }
}

ThreadGroupImpl::ThreadGroupImpl(StringPiece histogram_label,
                                 StringPiece thread_group_label,
                                 ThreadType thread_type_hint,
                                 TrackedRef<TaskTracker> task_tracker,
                                 TrackedRef<Delegate> delegate,
                                 ThreadGroup* predecessor_thread_group)
    : ThreadGroup(std::move(task_tracker),
                  std::move(delegate),
                  predecessor_thread_group),
      histogram_label_(histogram_label),
      thread_group_label_(thread_group_label),
      thread_type_hint_(thread_type_hint),
      idle_workers_set_cv_for_testing_(lock_.CreateConditionVariable()),
      tracked_ref_factory_(this) {
  DCHECK(!thread_group_label_.empty());
}

ThreadGroup::ThreadGroup(TrackedRef<TaskTracker> task_tracker,
                         TrackedRef<Delegate> delegate,
                         ThreadGroup* predecessor_thread_group)
    : task_tracker_(std::move(task_tracker)),
      delegate_(std::move(delegate)),
      lock_(predecessor_thread_group ? &predecessor_thread_group->lock_
                                     : nullptr) {
  DCHECK(task_tracker_);
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61

    线程池对象的实例化只是简单的创建了线程组对象，并创建线程组对象，一个线程可以管理多个线程组。chromium的线程组分为前台线程组，后台线程组和Utility线程组。投递任务的时候可以选择不同线程组。

    2、线程池的启动和线程的创建

    线程池的创建过程比较简单。 我们来看一下启动过程

    void ThreadPoolImpl::Start(const ThreadPoolInstance::InitParams& init_params,
                           WorkerThreadObserver* worker_thread_observer) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  DCHECK(!started_);

  size_t max_best_effort_tasks =
      std::min(kMaxBestEffortTasks, init_params.max_num_foreground_threads);

  ......
  // 1、先创建一个普通线程，作为service_thread, service_thread 用于控制执行顺序
  CHECK(service_thread_.StartWithOptions(std::move(service_thread_options)));
.....
  
  auto service_thread_task_runner = service_thread_.task_runner()；
  // 2、创建延时任务管理器， 注意这里传入的TaskRunner是使用service_thread_ 线程对象来创建的， 意图是把延时处理的逻辑放在单线程执行，来实现线程池内部的一些串行工作。
  delayed_task_manager_.Start(service_thread_task_runner);

  single_thread_task_runner_manager_.Start(service_thread_task_runner,
                                           worker_thread_observer);

 ......

  // On platforms that can't use the background thread priority, best-effort
  // tasks run in foreground pools. A cap is set on the number of best-effort
  // tasks that can run in foreground pools to ensure that there is always
  // room for incoming foreground tasks and to minimize the performance impact
  // of best-effort tasks.
  // 3、调用ThreadGroupImpl->Start() 启动线程组
  static_cast<ThreadGroupImpl*>(foreground_thread_group_.get())
      ->Start(foreground_threads, max_best_effort_tasks,
              init_params.suggested_reclaim_time, service_thread_task_runner,
              worker_thread_observer, worker_environment,
              g_synchronous_thread_start_for_testing);

 ......
  started_ = true;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37

    这里整体逻辑：
    1、先创建一个普通线程，作为service_thread_, 这里service_thread_的类型为Thread， 就是我们chromium线程模型(1)-普通线程实现（ui和io线程）这篇文章介绍的普通线程。 线程池对象持有一个普通线程对象作为service_thread_， 主要的目的是使用service_thread_ 单线程做一些串行化的工作。
    2、创建延时任务管理器delayed_task_manager_， 注意这里传入的TaskRunner是使用service_thread_ 线程对象来创建的， 意图是把延时处理的逻辑放在单线程执行，来实现串行。
    3、调用ThreadGroupImpl->Start() 启动线程组

    我们来具体看看线程组的启动

    src/base/task/thread_pool/thread_group_impl.cc

    void ThreadGroupImpl::Start(
    size_t max_tasks,
    size_t max_best_effort_tasks,
    TimeDelta suggested_reclaim_time,
    scoped_refptr<SingleThreadTaskRunner> service_thread_task_runner,
    WorkerThreadObserver* worker_thread_observer,
    WorkerEnvironment worker_environment,
    bool synchronous_thread_start_for_testing,
    absl::optional<TimeDelta> may_block_threshold) {
  ThreadGroup::Start();

  DCHECK(!replacement_thread_group_);

  in_start().no_worker_reclaim = FeatureList::IsEnabled(kNoWorkerThreadReclaim);
  in_start().may_block_threshold =
      may_block_threshold ? may_block_threshold.value()
                          : (thread_type_hint_ != ThreadType::kBackground
                                 ? kForegroundMayBlockThreshold
                                 : kBackgroundMayBlockThreshold);
  in_start().blocked_workers_poll_period =
      thread_type_hint_ != ThreadType::kBackground
          ? kForegroundBlockedWorkersPoll
          : kBackgroundBlockedWorkersPoll;

  ScopedCommandsExecutor executor(this);
  CheckedAutoLock auto_lock(lock_);

  DCHECK(workers_.empty());
  max_tasks_ = max_tasks;
  DCHECK_GE(max_tasks_, 1U);
  in_start().initial_max_tasks = max_tasks_;
  DCHECK_LE(in_start().initial_max_tasks, kMaxNumberOfWorkers);
  max_best_effort_tasks_ = max_best_effort_tasks;
  in_start().suggested_reclaim_time = suggested_reclaim_time;
  in_start().worker_environment = worker_environment;
  in_start().service_thread_task_runner = std::move(service_thread_task_runner);
  in_start().worker_thread_observer = worker_thread_observer;

.....

  EnsureEnoughWorkersLockRequired(&executor);
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43

    1、函数简单初始化一些配置参数

    • initial_max_tasks 初始化可以最大并发执行的任务数量
    • no_worker_reclaim： 该值为真表示线程idle时间太长不会被回收，只有多余线程才会回收（比initial_max_tasks多的空闲线程可以回收）。 否则所有线程都可以回收。
    • suggested_reclaim_time: 线程空闲多久需要回收
    • may_block_threshold： 执行MAY_BLOCK 任务的线程阻塞时间超过may_block_threshold 则认为线程真正发生了阻塞
    • blocked_workers_poll_period， 检查线程阻塞的周期，多久检查一次MAY_BLOCK的任务是否真正阻塞了线程。
    • worker_environment： 环境信息，只有在window系统有效
    • service_thread_task_runner: server_thread_的task_runner
    • worker_thread_observer: 用于监听工作线程创建和退出

    2、调用EnsureEnoughWorkersLockRequired(). 创建线程，EnsureEnoughWorkersLockRequired这个函数顾名思义是确保有足够多的工作者（也就是线程），所以现成的创建工作是在这里完成的，该函数也是线程数量维护的核心函数。后面我们会多次看到它。

    void ThreadGroupImpl::EnsureEnoughWorkersLockRequired(
    BaseScopedCommandsExecutor* base_executor) {
  ......
  ScopedCommandsExecutor* executor =
      static_cast<ScopedCommandsExecutor*>(base_executor);

  const size_t desired_num_awake_workers =
      GetDesiredNumAwakeWorkersLockRequired();
  const size_t num_awake_workers = GetNumAwakeWorkersLockRequired();

  size_t num_workers_to_wake_up =
      ClampSub(desired_num_awake_workers, num_awake_workers);
  // 1、计算需要唤醒idle线程的数量。
  num_workers_to_wake_up = std::min(num_workers_to_wake_up, size_t(2U));

  // Wake up the appropriate number of workers.
  //2 唤醒需要唤醒的线程
  for (size_t i = 0; i < num_workers_to_wake_up; ++i) {
    // 2.1 如果idle线程不够，创建线程，刚创建的线程是idle状态的，会放入到idle_workers_set_
    MaintainAtLeastOneIdleWorkerLockRequired(executor);
    // 2.2 从idle线程集合（idle_workers_set_）中拿出一个线程去唤醒
    WorkerThread* worker_to_wakeup = idle_workers_set_.Take();
    executor->ScheduleWakeUp(worker_to_wakeup);
  }
  ......
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    EnsureEnoughWorkersLockRequired函数首先根据任务情况计算需要唤醒的线程数量。然后按需唤醒线程。 唤醒线程分为2步：
    1、调用MaintainAtLeastOneIdleWorkerLockRequired函数，意图是当idle线程不足的时候创建新的线程，新的线程被创建出来之后默认是idle状态的，idle_workers_set_是idle状态线程的集合。
    2、从idle_workers_set_ 拿出一个idle线程去唤醒。唤醒线程调用ScopedCommandsExecutor->ScheduleWakeUp() 方法。

    我们先来分析MaintainAtLeastOneIdleWorkerLockRequired函数，这里有线程的创建过程。

    void ThreadGroupImpl::MaintainAtLeastOneIdleWorkerLockRequired(
    ScopedCommandsExecutor* executor) {
  if (workers_.size() == kMaxNumberOfWorkers) // 如果线程数量大于硬限制，则不再创建线程
    return;

  if (!idle_workers_set_.IsEmpty()) // 如果还有多余的idle线程，则不创建线程
    return;

  if (workers_.size() >= max_tasks_) // 大于最大软限制，则不再创建线程，max_tasks_ 可能根据系统任务负载情况调整，主要防止线程创建过快。
    return;

  // 调用CreateAndRegisterWorkerLockRequired 创建线程
  scoped_refptr<WorkerThread> new_worker =
      CreateAndRegisterWorkerLockRequired(executor);
  // 添加到idle集合
  idle_workers_set_.Insert(new_worker.get());
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    MaintainAtLeastOneIdleWorkerLockRequired ，先判断是否能够继续创建线程， 判断条件包
    1、括线程数量是否超出了系统的硬限制kMaxNumberOfWorkers 和软限制ThreadGroupImpl.max_tasks_。
    2、是否还有多余的idle线程。

    如果需要创建线程则调用CreateAndRegisterWorkerLockRequired(executor) 函数进行创建。线程用WorkerThread对象描述，线程初始状态为idle状态，所以放入idle_workers_set_集合。

    我们继续分析

    
scoped_refptr<WorkerThread>
ThreadGroupImpl::CreateAndRegisterWorkerLockRequired(
    ScopedCommandsExecutor* executor) {
 .......

  // 创建线程对象WorkerThread
  scoped_refptr<WorkerThread> worker = MakeRefCounted<WorkerThread>(
      thread_type_hint_,
      std::make_unique<WorkerThreadDelegateImpl>(
          tracked_ref_factory_.GetTrackedRef(),
          /* is_excess=*/after_start().no_worker_reclaim
              ? workers_.size() >= after_start().initial_max_tasks
              : true),
      task_tracker_, worker_sequence_num_++, &lock_);
  // 先加到全部工作线程集合， workers_包括runing 和idle的线程
  workers_.push_back(worker);
  
  // 调度启动线程
  executor->ScheduleStart(worker);
  return worker;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    函数创建WorkerThread 实例作为线程实例，然后添加到全部线程集合ThreadGroupImpl.workers_中， 最后调用executor->ScheduleStart(worker)启动线程。

    WorkerThread创建比较简单，我们来看启动

      void ThreadGroupImpl::ScopedCommandsExecutor::ScheduleStart(scoped_refptr<WorkerThread> worker) {
    workers_to_start_.AddWorker(std::move(worker));
  }

    • 1
    • 2
    • 3

    executor是ThreadGroupImpl::ScopedCommandsExecutor的实例，在EnsureEnoughWorkersLockRequired函数的时候创建。chromium中叫Scopedxxx的类一般都在作用域内获取资源，在作用域技术后对象析构函数做一些额外工作。

    ScopedCommandsExecutor类也是这样的对象。我们前面还看到了唤醒线程的工作

      void ThreadGroupImpl::ScopedCommandsExecutor::ScheduleWakeUp(scoped_refptr<WorkerThread> worker) {
    workers_to_wake_up_.AddWorker(std::move(worker));
  }


    • 1
    • 2
    • 3
    • 4

    我们可以看到线程启动是把WorkerThread实例放到了workers_to_start_ 集合， 而线程唤醒是把WorkerThread实例放到了workers_to_wake_up_集合中。 我们可以想到ScopedCommandsExecutor的作用是把任务先收集起来，到作用域结束调用ScopedCommandsExecutor析构方法的时候批量执行线程启动和唤醒。

    ~ScopedCommandsExecutor() { FlushImpl(); }

    • 1

    ScopedCommandsExecutor 析构后调用 FlushImpl() 函数执行批量提操作

      void FlushImpl() {
    CheckedLock::AssertNoLockHeldOnCurrentThread();

    // 唤醒需要唤醒的线程
    workers_to_wake_up_.ForEachWorker(
        [](WorkerThread* worker) { worker->WakeUp(); });

    // 启动需要启动的线程
    workers_to_start_.ForEachWorker([&](WorkerThread* worker) {
      worker->Start(outer_->after_start().service_thread_task_runner,
                    outer_->after_start().worker_thread_observer);
      ......
    });

    ......
  }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    FlushImpl 函数调用了WorkerThread的Start方法启动线程，调用WorkerThread->WakeUp() 函数唤醒线程。
    先来看线程的启动

    bool WorkerThread::Start(
    scoped_refptr<SingleThreadTaskRunner> io_thread_task_runner,
    WorkerThreadObserver* worker_thread_observer) {
  CheckedLock::AssertNoLockHeldOnCurrentThread();

......

  CheckedAutoLock auto_lock(thread_lock_);
  #if (BUILDFLAG(IS_POSIX) && !BUILDFLAG(IS_NACL)) || BUILDFLAG(IS_FUCHSIA)
  DCHECK(io_thread_task_runner);
  io_thread_task_runner_ = std::move(io_thread_task_runner);
#endif
......
  // 设置了线程需要结束，或者调用了join， 直接返回true
  if (should_exit_.IsSet() || join_called_for_testing_.IsSet()) 
    return true;

  DCHECK(!worker_thread_observer_);
  worker_thread_observer_ = worker_thread_observer;

  self_ = this;

  constexpr size_t kDefaultStackSize = 0;
  // 创建平台线程
  PlatformThread::CreateWithType(kDefaultStackSize, this, &thread_handle_,
                                 current_thread_type_);

  if (thread_handle_.is_null()) {
    self_ = nullptr;
    return false;
  }

  return true;
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35

    这里把WorkerThread的io_thread_task_runner_ 设置为ThreadPool的service_thread_创建的TaskRunner，后面会用到service_thread_ 线程做一些串行化操作。
    创建平台线程PlatformThread，在linux上为pthread， pthread启动后会在线程中调用WorkerThread的ThreadMain() 方法。注意这里执行线程已经转到了WorkerThread线程内。

    void WorkerThread::ThreadMain() {
// 把文件观察期（io事件观察）的TaskRunner 设置为io_thread_task_runner_。 也就是在service_thread_监听文件io事件。
FileDescriptorWatcher file_descriptor_watcher(io_thread_task_runner_);

  if (thread_type_hint_ == ThreadType::kBackground) {
    switch (delegate_->GetThreadLabel()) {
      case ThreadLabel::POOLED:
        RunBackgroundPooledWorker();
        return;
      case ThreadLabel::SHARED:
        RunBackgroundSharedWorker();
        return;
      case ThreadLabel::DEDICATED:
        RunBackgroundDedicatedWorker();
        return;
#if BUILDFLAG(IS_WIN)
      case ThreadLabel::SHARED_COM:
        RunBackgroundSharedCOMWorker();
        return;
      case ThreadLabel::DEDICATED_COM:
        RunBackgroundDedicatedCOMWorker();
        return;
#endif  // BUILDFLAG(IS_WIN)
    }
  }

  switch (delegate_->GetThreadLabel()) {
    case ThreadLabel::POOLED:
      RunPooledWorker();
      return;
    case ThreadLabel::SHARED:
      RunSharedWorker();
      return;
    case ThreadLabel::DEDICATED:
      RunDedicatedWorker();
      return;
#if BUILDFLAG(IS_WIN)
    case ThreadLabel::SHARED_COM:
      RunSharedCOMWorker();
      return;
    case ThreadLabel::DEDICATED_COM:
      RunDedicatedCOMWorker();
      return;
#endif  // BUILDFLAG(IS_WIN)
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46

    这里根据线程组和线程的不同类型执行相应的线程循环。我们前面关注的线程组类型是kDefault， ThreadLabel::POOLED，所以我们分析RunPooledWorker()函数 ，其他类型的线程循环读者可进行自行分析。

    NOINLINE void WorkerThread::RunPooledWorker() {
  RunWorker();
  NO_CODE_FOLDING();
}

void WorkerThread::RunWorker() {
  ......
  
  delegate_->OnMainEntry(this);
  ......

  // 线程默认为idle状态。等待在唤醒条件变量上（wake_up_event_）
  {
    TRACE_EVENT_END0("base", "WorkerThread active");
    // TODO(crbug.com/1021571): Remove this once fixed.
    PERFETTO_INTERNAL_ADD_EMPTY_EVENT();
    delegate_->WaitForWork(&wake_up_event_);
    TRACE_EVENT_BEGIN("base", "WorkerThread active",
                      perfetto::TerminatingFlow::FromPointer(this));
  }
  bool got_work_this_wakeup = false;
  while (!ShouldExit()) {
    ......
    // 1、 获取任务源
    RegisteredTaskSource task_source = delegate_->GetWork(this);
    .......

    // Alias pointer for investigation of memory corruption. crbug.com/1218384
    TaskSource* task_source_before_run = task_source.get();
    base::debug::Alias(&task_source_before_run);
    // 从任务源获取任务并执行
    task_source = task_tracker_->RunAndPopNextTask(std::move(task_source));
    // 任务源需要重新入队列，清除一些执行过程中设置的信息
    delegate_->DidProcessTask(std::move(task_source));

  ......
    wake_up_event_.Reset();
  }

  // Important: It is unsafe to access unowned state (e.g. |task_tracker_|)
  // after invoking OnMainExit().

  delegate_->OnMainExit(this);

 ......
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47

    消息的循环总体比较简单， 就是一个死循环，通过delegate_->GetWork(this) 获取可执行的任务源， 并执行任务。 这里我们可以看到任务idle时是调用delegate_->WaitForWork(&wake_up_event_) 等待唤醒。 我们再看下线程唤醒逻辑

    void WorkerThread::WakeUp() {
  ......
  wake_up_event_.Signal();
}

    • 1
    • 2
    • 3
    • 4

    通过wake_up_event_ 条件变量唤醒。

    好了我们继续分析任务的获取和执行逻辑。delegate_ 为ThreadGroupImpl::WorkerThreadDelegateImpl类的实例。

    任务的获取和执行

    
RegisteredTaskSource ThreadGroupImpl::WorkerThreadDelegateImpl::GetWork(
    WorkerThread* worker) {
 
  ScopedCommandsExecutor executor(outer_.get());
  CheckedAutoLock auto_lock(outer_->lock_);


  // 执行EnsureEnoughWorkersLockRequired() 根据需求创建或者唤醒额外线程，这个函数我们前面分析过了
  outer_->EnsureEnoughWorkersLockRequired(&executor);
  executor.FlushWorkerCreation(&outer_->lock_);
  // 尝试回收work线程，如果成功回收此线程不能执行该任务。如果唤醒的线程数量超过总任务，该线程需要进入idle状态，该线程也不能执行任务，直接返回nullprr
  if (!CanGetWorkLockRequired(&executor, worker))
    return nullptr;

  RegisteredTaskSource task_source;
  TaskPriority priority;
  while (!task_source && !outer_->priority_queue_.IsEmpty()) {
    // Enforce the CanRunPolicy and that no more than |max_best_effort_tasks_|
    // BEST_EFFORT tasks run concurrently.
    // 如果最高优先级任务不能运行，放弃查找
    priority = outer_->priority_queue_.PeekSortKey().priority();
    if (!outer_->task_tracker_->CanRunPriority(priority) ||
        (priority == TaskPriority::BEST_EFFORT &&
         outer_->num_running_best_effort_tasks_ >=
             outer_->max_best_effort_tasks_)) {
      break;
    }
    // 任务可以运行。获取任务源
    task_source = outer_->TakeRegisteredTaskSource(&executor);
  }
  if (!task_source) { 
    // 没有获取到任务。 设置线程为idle状态
    OnWorkerBecomesIdleLockRequired(&executor, worker);
    return nullptr;
  }

  // Running task bookkeeping.
  // 增加任务运行的计数
  outer_->IncrementTasksRunningLockRequired(priority);
 ......

  return task_source;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44

    函数按照一下几个步骤执行:

    1. 执行EnsureEnoughWorkersLockRequired() 根据需求创建或者唤醒额外线程，这个函数我们前面分析过了, 注释说这里是为了降低开销，因为这里持有锁，比在PostTask时候执行开销要小。

    2. 尝试回收work线程，如果成功回收此线程不能执行该任务。如果唤醒的线程数量超过总任务，该线程需要进入idle状态，该线程也不能执行任务，直接返回nullptr。 此为线程回收和状态管理的一部分，我们后面分析。

    3. 从任务优先级队列获取任务源。
      3.1 如果当前优先级队列里面的最高优先级任务源不能执行，则无法获取可执行任务源
      3.2 如果最高优先级任务源可以执行，但是优先级是BEST_EFFORT，但是正在运行的BEST_EFFORT优先级任务已经超出最大限制（max_best_effort_tasks_），则也无法获取有效任务源。
      3.3 经过前两步检查，调用ThreadGroupImpl->TakeRegisteredTaskSource() 获取任务源。

    4. 检查获取到的任务源，如果为空则该表示没有任务可以执行，设置当前线程状态为idle。

    5. 调用ThreadGroupImpl->IncrementTasksRunningLockRequired(() 增加正在运行的任务计数。

    获取任务源我们分析完了，再来分析下如何运行任务源里面的任务

    RegisteredTaskSource TaskTracker::RunAndPopNextTask(
    RegisteredTaskSource task_source) {
  DCHECK(task_source);

  const bool should_run_tasks = BeforeRunTask(task_source->shutdown_behavior());

  // Run the next task in |task_source|.
  absl::optional<Task> task;
  TaskTraits traits;
  {
    // 开始事务，其实就是获取相关锁
    auto transaction = task_source->BeginTransaction();
    // 如果shatdown状态允许执行任务，就调用TaskTask获取一个任务，否则清空任务，返回一个假任务。
    task = should_run_tasks ? task_source.TakeTask(&transaction)
                            : task_source.Clear(&transaction);
    traits = transaction.traits();
  }

  if (task) {
    // Skip delayed tasks if shutdown started.
    // 延期任务，并且已经开始关闭chromium状态，则构造一个假任务
    if (!task->delayed_run_time.is_null() && state_->HasShutdownStarted())
      task->task = base::DoNothingWithBoundArgs(std::move(task->task));

    // Run the |task| (whether it's a worker task or the Clear() closure).
    // 执行任务
    RunTask(std::move(task.value()), task_source.get(), traits);
  }
  if (should_run_tasks)
    AfterRunTask(task_source->shutdown_behavior());
  const bool task_source_must_be_queued = task_source.DidProcessTask();
  // |task_source| should be reenqueued iff requested by DidProcessTask().
  if (task_source_must_be_queued)
    return task_source;
  return nullptr;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36

    函数总体比较简单，自行看注释，函数主要从TaskSource里获取任务并执行，并调用RunTask执行任务。

    线程的回收
    前面我们看到了Workerthread 没有空闲的时候休眠在WorkerThread->wake_up_event_ 条件变量上面。任务创建的时候就是idle状态，放在ThreadGroupImpl->idle_workers_set_ 里面，在EnsureEnoughWorkersLockRequired的时候按需唤醒idle线程

    void ThreadGroupImpl::EnsureEnoughWorkersLockRequired(
    BaseScopedCommandsExecutor* base_executor) {
 ......
  // Wake up the appropriate number of workers.
  for (size_t i = 0; i < num_workers_to_wake_up; ++i) {
    MaintainAtLeastOneIdleWorkerLockRequired(executor);
    WorkerThread* worker_to_wakeup = idle_workers_set_.Take();
    DCHECK(worker_to_wakeup);
    executor->ScheduleWakeUp(worker_to_wakeup);
  }
......
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    先来看下任务是如何进入idle状态的。首先是任务创建的时候，直接进入idle状态。idle状态的线程都在idle_workers_set_集合中， 在唤醒的时候变成非idle状态，WorkerThread有一个last_used_time_成员变量，记录thread最后一次使用时间

    idle_workers_set_.Insert(new_worker.get())

void WorkerThreadSet::Insert(WorkerThread* worker) {
  DCHECK(!Contains(worker)) << "WorkerThread already on stack";
  auto old_first = set_.begin();
  set_.insert(worker);
  if (worker != *set_.begin())
    worker->BeginUnusedPeriod();
  else if (old_first != set_.end())
    (*old_first)->BeginUnusedPeriod();
}
WorkerThread* WorkerThreadSet::Take() {
  if (IsEmpty())
    return nullptr;
  WorkerThread* const worker = *set_.begin();
  set_.erase(set_.begin());
  if (!IsEmpty())
    (*set_.begin())->EndUnusedPeriod();
  return worker;
}

void WorkerThread::BeginUnusedPeriod() {
  CheckedAutoLock auto_lock(thread_lock_);
  DCHECK(last_used_time_.is_null());
  // 设置为当前时间
  last_used_time_ = subtle::TimeTicksNowIgnoringOverride();
}

void WorkerThread::EndUnusedPeriod() {
  CheckedAutoLock auto_lock(thread_lock_);
  DCHECK(!last_used_time_.is_null());
  // 设置为0
  last_used_time_ = TimeTicks();
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34

    我们可以看到，idle_workers_set_ 里面的第一个WorkerThread->last_used_time_ 为0， 其他WorkerThread->last_used_time_为进入idle_workers_set_的时间。 获取idle_work 都是从idle_workers_set_第一个元素开始。这样出队列后不需要设置WorkerThread->last_used_time_ 为0。

    任务首次进入idle_workers_set_ 的代码我们已经分析过了。另外一种情况就是工作线程数量大于总任务源数量。这个逻辑在ThreadGroupImpl::WorkerThreadDelegateImpl::GetWork() 函数中，也就是获取要执行的任务时

    RegisteredTaskSource ThreadGroupImpl::WorkerThreadDelegateImpl::GetWork(
    WorkerThread* worker) {
  ......
  if (!CanGetWorkLockRequired(&executor, worker))
    return nullptr;

 ......
  if (!task_source) {
    OnWorkerBecomesIdleLockRequired(&executor, worker);
    return nullptr;
  }

......

  return task_source;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    当没有可以执行的任务源时候调用OnWorkerBecomesIdleLockRequired(&executor, worker) 使线程进入idle状态

    void ThreadGroupImpl::WorkerThreadDelegateImpl::OnWorkerBecomesIdleLockRequired(
    ScopedCommandsExecutor* executor,
    WorkerThread* worker) {
......
  // Add the worker to the idle set.
  outer_->idle_workers_set_.Insert(worker);
......
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    另外CanGetWorkLockRequired(&executor, worker) 函数也会判断任务是否需要进入idle状态。

    bool ThreadGroupImpl::WorkerThreadDelegateImpl::CanGetWorkLockRequired(
    ScopedCommandsExecutor* executor,
    WorkerThread* worker) {
  // IsOnIdleSetLockRequired() 判断WorkerThread 是否在idle_workers_set_ 中
  const bool is_on_idle_workers_set = outer_->IsOnIdleSetLockRequired(worker);
......
  // 如果在idle work中， 尝试回收任务，说明这次唤醒是需要回收而唤醒的
  if (is_on_idle_workers_set) {
    if (CanCleanupLockRequired(worker))
      CleanupLockRequired(executor, worker);
    return false;
  }

  // 没有回收，线程数量超过最大并发，调用OnWorkerBecomesIdleLockRequired 回收
  if (outer_->GetNumAwakeWorkersLockRequired() > outer_->max_tasks_) {
    OnWorkerBecomesIdleLockRequired(executor, worker);
    return false;
  }

  return true;
}


bool ThreadGroupImpl::IsOnIdleSetLockRequired(WorkerThread* worker) const {
  // To avoid searching through the idle set : use GetLastUsedTime() not being
  // null (or being directly on top of the idle set) as a proxy for being on
  // the idle set.
  return idle_workers_set_.Peek() == worker ||
         !worker->GetLastUsedTime().is_null();
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30

    这里先判断任务是否在idle_workers_set_中，如果是则说明这次唤醒可能是为了回收任务， 尝试回收。 如果，没有回收，说明线程在idle_workers_set_中，那么如果唤醒线程数超过最大并发数，调用OnWorkerBecomesIdleLockRequired 使任务进入idle。（为任务唤醒的时候不会给idle_workers_set_里面的线程发送信号）。

    这里我们也看到了线程回收的函数CanCleanupLockRequired。

    bool ThreadGroupImpl::WorkerThreadDelegateImpl::CanCleanupLockRequired(
    const WorkerThread* worker) const {
  DCHECK_CALLED_ON_VALID_THREAD(worker_thread_checker_);
  if (!is_excess())
    return false;

  const TimeTicks last_used_time = worker->GetLastUsedTime();
  return !last_used_time.is_null() &&
         subtle::TimeTicksNowIgnoringOverride() - last_used_time >=
             outer_->after_start().suggested_reclaim_time &&
         LIKELY(!outer_->worker_cleanup_disallowed_for_testing_);
}



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    CanCleanupLockRequired函数是线程是否可以回收的条件， 也就是当前时间-进入idle的时间 如果大于建议回收的时间，则可以回收，其实就是空闲时间超过suggested_reclaim_time（outer_->worker_cleanup_disallowed_for_testing_ 为单元测试的条件，一般为假）。

    void ThreadGroupImpl::WorkerThreadDelegateImpl::CleanupLockRequired(
    ScopedCommandsExecutor* executor,
    WorkerThread* worker) {
  DCHECK(!outer_->join_for_testing_started_);
  DCHECK_CALLED_ON_VALID_THREAD(worker_thread_checker_);

  worker->Cleanup();

  if (outer_->IsOnIdleSetLockRequired(worker))
    outer_->idle_workers_set_.Remove(worker);

  // Remove the worker from |workers_|.
  auto worker_iter = ranges::find(outer_->workers_, worker);
  DCHECK(worker_iter != outer_->workers_.end());
  outer_->workers_.erase(worker_iter);
}

void WorkerThread::Cleanup() {
  DCHECK(!should_exit_.IsSet());
  should_exit_.Set();
  wake_up_event_.Signal();
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    CleanupLockRequired 为线程回收的函数。 主要设置WorkerThread.should_exit_ 状态，并且唤醒线程。在下一次线程循环的时候线程就正常退出了。 并且从ThreadGroupImpl->idle_workers_set_ 和 ThreadGroupImpl->workers_ 中删除WorkerThread。

    我们可以看到，WorkerThread 空闲时休眠在自己持有的条件变量上面。那么一个线程进已经进入idle状态就不会给他分配任务，要回收它就需要由外部唤醒， 何时进行唤醒呢？

    我们来看WorkerThread是如何等待在条件变量上的

    void WorkerThread::Delegate::WaitForWork(WaitableEvent* wake_up_event) {
  DCHECK(wake_up_event);
  const TimeDelta sleep_time = GetSleepTimeout();
  .......
  wake_up_event->TimedWait(sleep_time);
......
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    原来等待条件变量有一个超时时间，超过这个时间就会自动唤醒， 我们看下超时时间是如何设置的。

    TimeDelta ThreadGroupImpl::WorkerThreadDelegateImpl::GetSleepTimeout() {
  DCHECK_CALLED_ON_VALID_THREAD(worker_thread_checker_);
  if (!is_excess())
    return TimeDelta::Max();
  ......
  return outer_->after_start().suggested_reclaim_time * 1.1;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    如果线程是不可回收的， 那么超时时间设置的非常大。 否则设置的超时时间为suggested_reclaim_time， 该值是创建线程组的时候传递过来的，建议的线程空闲回收时间。 这样我们就弄明白了线程idle 和回收逻辑。

    任务数量的增长
    一开始的时候我们说过线程池线程数量会随着需求增长，我们来分析下增长逻辑。
    ThreadGroupImpl.max_tasks_ 值代表可以创建的最大线程数， 我们说过有一些io相关的工作，这类型的工作会使线程阻塞，但是不会占用太多的cpu资源， 所以遇到这种线程的时候可以临时提高最大线程数， 当这些工作处理完成之后再降低线程数，这样不会增加太多系统负载，还能不耽误非阻塞任务的运行（这就是没有协程的麻烦之处）。

    在分析程序之前我们来介绍几个变量：
    ThreadGroupImpl.max_tasks_：最大可以创建的线程数
    ThreadGroupImpl.max_best_effort_tasks_：最大运行的BEST_EFFORT优先级的线程数
    ThreadGroupImpl.num_running_best_effort_tasks_: 表示正在运行的BEST_EFFORT 优先级的线程数。
    ThreadGroupImpl.num_running_tasks_: 正在运行的所有线程数，包含num_running_best_effort_tasks_
    ThreadGroupImpl.num_unresolved_may_block_： 前面我们说了chromium把阻塞类型分为可能阻塞MAY_BLOCK， 和一定阻塞WILL_BLOCK， MAY_BLOCK在阻塞前会不会真正阻塞不确定，未确定的正在运行的阻塞任务称为unresolved_may_block_任务，所以ThreadGroupImpl.num_unresolved_may_block_表示正在运行的unresolved_may_block_任务个数。
    ThreadGroupImpl.num_unresolved_best_effort_may_block_ 表示优先级是BEST_EFFORT 的unresolved_may_block_任务个数。

    好了了解到这些变量之后我们来分析线程增加的逻辑。
    我们前面看见可以可以继续创建线程的依据是ThreadGroupImpl.max_tasks_

    void ThreadGroupImpl::MaintainAtLeastOneIdleWorkerLockRequired(
    ScopedCommandsExecutor* executor) {
  if (workers_.size() == kMaxNumberOfWorkers)
    return;
......
  if (workers_.size() >= max_tasks_)
    return;

 .......
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    这段代码可以看出，除了一个硬限制，线程池维护的线程数量主要参考ThreadGroupImpl.max_tasks_ 值。那我们主要关注ThreadGroupImpl.max_tasks_ 的变化。
    另外用户阻塞任务执行之前需要提示线程池该任务是阻塞任务，并且告知线程池阻塞类型是MAY_BLOCK 还是WILL_BLOCK， 要怎么做呢, 在base/threading/scoped_blocking_call.h 头文件里面提供了例子。

       {
     ScopedBlockingCall scoped_blocking_call(
        FROM_HERE, BlockingType::WILL_BLOCK);
     data = GetDataFromNetwork();
   }
   CPUIntensiveProcessing(data);

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    也就是在阻塞任务开始前创建ScopedBlockingCall， 结束后析构ScopedBlockingCall。 ScopedBlockingCall创建后会调用ThreadGroupImpl::WorkerThreadDelegateImpl::BlockingStarted()， 这个函数是在WorkerThread线程调用的

    void ThreadGroupImpl::WorkerThreadDelegateImpl::BlockingStarted(
    BlockingType blocking_type) {
  // 不是在运行任务的时候调用的，直接返回， 因为不能保证该线程是WorkerThread对应线程， 运行任务的时候会设置current_task_priority
  if (!read_worker().current_task_priority) {
    return;
  }
 ......

  ScopedCommandsExecutor executor(outer_.get());
  CheckedAutoLock auto_lock(outer_->lock_);

 ......
  // 设置线程阻塞开始时间，用于检测是否真正发生阻塞
  write_worker().blocking_start_time = subtle::TimeTicksNowIgnoringOverride();

 ......
  // 如果优先级是BEST_EFFORT， 增加num_unresolved_best_effort_may_block_ 计数
  if (*read_any().current_task_priority == TaskPriority::BEST_EFFORT)
    ++outer_->num_unresolved_best_effort_may_block_;

  if (blocking_type == BlockingType::WILL_BLOCK) {
    // 如果线程直接是WILL_BLOCK则直接增加ThreadGroupImpl.max_tasks_, incremented_max_tasks_since_blocked_ 用于记录是否由于任务增加郭max_tasks_， 防止重复增加
    incremented_max_tasks_since_blocked_ = true;
    outer_->IncrementMaxTasksLockRequired();
    outer_->EnsureEnoughWorkersLockRequired(&executor);
  } else {
    // 增加num_unresolved_may_block_ 
    ++outer_->num_unresolved_may_block_;
  }
  // 启动检测任务阻塞
  outer_->MaybeScheduleAdjustMaxTasksLockRequired(&executor);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32

    函数分为4部分
    1、检查是不是在运行任务的时候调用的，如果不是直接返回
    2、设置线程阻塞开始时间 write_worker().blocking_start_time，用于检测是否真正发生阻塞
    3、如果优先级是BEST_EFFORT， 增加num_unresolved_best_effort_may_block_ 计数， 说明BEST_EFFORT优先级线程有进一步抑制。
    4、如果线程直接是WILL_BLOCK则直接增加ThreadGroupImpl.max_tasks_, incremented_max_tasks_since_blocked_ 用于记录是否由于任务增加郭max_tasks_， 防止重复增加max_task_，也就是一个被阻塞的线程最多增加一个线程补充。 如果线程不是WILL_BLOCK的，还不能确定线程是否阻塞，则增加num_unresolved_may_block_.
    5、调用MaybeScheduleAdjustMaxTasksLockRequired函数启动检测任务阻塞。

    void ThreadGroupImpl::MaybeScheduleAdjustMaxTasksLockRequired(
    ScopedCommandsExecutor* executor) {
  if (!adjust_max_tasks_posted_ &&
      ShouldPeriodicallyAdjustMaxTasksLockRequired()) {
    executor->ScheduleAdjustMaxTasks();
    adjust_max_tasks_posted_ = true;
  }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    adjust_max_tasks_posted_ 表示已经启动过阻塞检测任务了。就不需要再次启动，ShouldPeriodicallyAdjustMaxTasksLockRequired()条件则进一步检测是否需要启动阻塞任务检测。如果确实需要启动就调用executor->ScheduleAdjustMaxTasks()启动检测。我们先来看下判断条件

    bool ThreadGroupImpl::ShouldPeriodicallyAdjustMaxTasksLockRequired() {

   ......
  // 获取理想的BEST_EFFORT 优先级线程个数
  const size_t num_running_or_queued_best_effort_task_sources =
      num_running_best_effort_tasks_ +
      GetNumAdditionalWorkersForBestEffortTaskSourcesLockRequired();
  // 如果理想的BEST_EFFORT 优先级线程个数已经超过max_best_effort_tasks_(最大BEST_EFFORT 可运行个数)， 并且有未确认阻塞的BEST_EFFORT 任务，就需要尝试提升max_best_effort_tasks_， 来加快BEST_EFFORT 优先级任务执行。 BEST_EFFORT 线程因为可能被阻塞占用了
  if (num_running_or_queued_best_effort_task_sources > max_best_effort_tasks_ &&
      num_unresolved_best_effort_may_block_ > 0) {
    return true;
  }
  // 如果理想需要的线程个数已经超过max_tasks_， 并且有未确认阻塞的任务，就需要尝试提升max_tasks_， 来加快任务执行。 线程因为可能被阻塞占用了
  const size_t num_running_or_queued_task_sources =
      num_running_tasks_ +
      GetNumAdditionalWorkersForBestEffortTaskSourcesLockRequired() +
      GetNumAdditionalWorkersForForegroundTaskSourcesLockRequired();
  constexpr size_t kIdleWorker = 1;
  return num_running_or_queued_task_sources + kIdleWorker > max_tasks_ &&
         num_unresolved_may_block_ > 0;
}

size_t
ThreadGroup::GetNumAdditionalWorkersForBestEffortTaskSourcesLockRequired()
    const {
 
  //BEST_EFFORT 优先级的任务源个数
  const size_t num_queued =
      priority_queue_.GetNumTaskSourcesWithPriority(TaskPriority::BEST_EFFORT);
  // 如果任务源是0，或者不允许运行BEST_EFFORT优先级任务，返回0
  if (num_queued == 0 ||
      !task_tracker_->CanRunPriority(TaskPriority::BEST_EFFORT)) {
    return 0U;
  }
  // 如果优先级队列里面第一个任务优先级是BEST_EFFORT，则获取该任务源提示的并发度。 总计根据任务源算出来的并发度为  （该优先级任务个数-1） + 第一个源需要的并发任务个数
  if (priority_queue_.PeekSortKey().priority() == TaskPriority::BEST_EFFORT) {
    // Assign the correct number of workers for the top TaskSource (-1 for the
    // worker that is already accounted for in |num_queued|).
    return std::max<size_t>(
        1, num_queued +
               priority_queue_.PeekTaskSource()->GetRemainingConcurrency() - 1);
  }
  return num_queued;
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45

    如果理想的BEST_EFFORT 优先级线程个数已经超过max_best_effort_tasks_(最大BEST_EFFORT 可运行个数)， 并且有未确认阻塞的BEST_EFFORT 任务，就需要尝试提升max_best_effort_tasks_， 来加快BEST_EFFORT 优先级任务执行。 BEST_EFFORT 线程因为可能被阻塞占用了。
    如果理想需要的线程个数已经超过max_tasks_， 并且有未确认阻塞的任务，就需要尝试提升max_tasks_， 来加快任务执行。 线程因为可能被阻塞占用了。
    总结一下就是当线程比较紧张的时候又有不确定阻塞的任务时发起检测。

    
  void ThreadGroupImpl::ScopedCommandsExecutor::ScheduleAdjustMaxTasks() {
    DCHECK(!must_schedule_adjust_max_tasks_);
    must_schedule_adjust_max_tasks_ = true;
  }
  void FlushImpl() {
    ......

    if (must_schedule_adjust_max_tasks_)
      outer_->ScheduleAdjustMaxTasks();
  }

void ThreadGroupImpl::ScheduleAdjustMaxTasks() {
  ......
  after_start().service_thread_task_runner->PostDelayedTask(
      FROM_HERE, BindOnce(&ThreadGroupImpl::AdjustMaxTasks, Unretained(this)),
      after_start().blocked_workers_poll_period);
 ......
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    该函数最终会发送延时任务，执行ThreadGroupImpl::AdjustMaxTasks 方法，用于检测任务是否真正阻塞，从而提升任务数量。

    
void ThreadGroupImpl::AdjustMaxTasks() {
  DCHECK(
      after_start().service_thread_task_runner->RunsTasksInCurrentSequence());

  ScopedCommandsExecutor executor(this);
 ......
 // 设置adjust_max_tasks_posted_ 以便可以再次执行检测
  adjust_max_tasks_posted_ = false;

  // 对所有WorkerThread执行MaybeIncrementMaxTasksLockRequired(), 用于确定是否阻塞，和提升线程个数
  for (scoped_refptr<WorkerThread> worker : workers_) {
    // The delegates of workers inside a ThreadGroupImpl should be
    // WorkerThreadDelegateImpls.
    WorkerThreadDelegateImpl* delegate =
        static_cast<WorkerThreadDelegateImpl*>(worker->delegate());
    AnnotateAcquiredLockAlias annotate(lock_, delegate->lock());
    delegate->MaybeIncrementMaxTasksLockRequired();
  }

  // 增加线程
  EnsureEnoughWorkersLockRequired(&executor);
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    1、设置adjust_max_tasks_posted_ 以便可以再次执行检测
    2、对所有WorkerThread执行MaybeIncrementMaxTasksLockRequired(), 用于确定是否阻塞，和提升线程个数
    3、调用 EnsureEnoughWorkersLockRequired() 用于创建线程。

    void ThreadGroupImpl::WorkerThreadDelegateImpl::
    MaybeIncrementMaxTasksLockRequired() {
    // 线程阻塞时间超过了may_block_threshold 表示确实阻塞
  if (read_any().blocking_start_time.is_null() ||
      subtle::TimeTicksNowIgnoringOverride() - read_any().blocking_start_time <
          outer_->after_start().may_block_threshold) {
    return;
  }
  // 如果真的发生阻塞调用IncrementMaxTasksLockRequired()
  IncrementMaxTasksLockRequired();
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    函数先检测线程阻塞时间超过了may_block_threshold 表示确实阻塞。 /如果真的发生阻塞调用IncrementMaxTasksLockRequired() 增加ThreadGroupImpl->max_tasks。

    void ThreadGroupImpl::WorkerThreadDelegateImpl::
    IncrementMaxTasksLockRequired() {
  if (!incremented_max_tasks_since_blocked_) { // 判断是否已经为该阻塞线程补充了线程
    // 增加最大线程数ThreadGroupImpl.max_tasks_，减少num_unresolved_may_block_
    outer_->IncrementMaxTasksLockRequired();
    // Update state for an unresolved ScopedBlockingCall.
    if (!read_any().blocking_start_time.is_null()) {
      incremented_max_tasks_since_blocked_ = true;
      --outer_->num_unresolved_may_block_;
    }
  }
  if (*read_any().current_task_priority == TaskPriority::BEST_EFFORT &&
      !incremented_max_best_effort_tasks_since_blocked_) { // 判断是否已经为该阻塞线程补充了线程
     // 增加最大线程数ThreadGroupImpl.max_best_effort_tasks_，减少num_unresolved_may_block_
    outer_->IncrementMaxBestEffortTasksLockRequired(); 
    // Update state for an unresolved ScopedBlockingCall.
    if (!read_any().blocking_start_time.is_null()) {
      incremented_max_best_effort_tasks_since_blocked_ = true;
      --outer_->num_unresolved_best_effort_may_block_;
    }
  }
}

void ThreadGroupImpl::DecrementMaxBestEffortTasksLockRequired() {
  DCHECK_GT(num_running_tasks_, 0U);
  DCHECK_GT(max_best_effort_tasks_, 0U);
  --max_best_effort_tasks_;
  UpdateMinAllowedPriorityLockRequired();
}

void ThreadGroupImpl::IncrementMaxBestEffortTasksLockRequired() {
  DCHECK_GT(num_running_tasks_, 0U);
  ++max_best_effort_tasks_;
  UpdateMinAllowedPriorityLockRequired();
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36

    函数判断如果没有为该已经阻塞的线程补充郭线程，就调用对应方法增加ThreadGroupImpl->max_tasks_ ，如果该任务优先级是BEST_EFFORT 还要增加
    ThreadGroupImpl->max_best_effort_tasks_计数。 这里我们也可以看出BEST_EFFORT 优先级的阻塞任务会同时增加ThreadGroupImpl->max_best_effort_tasks_ 和 ThreadGroupImpl->max_tasks_ 。这也可以说明BEST_EFFORT 优先级任务并行执行的个数是有进一步限制的。

    我们再来分析一下如何减少ThreadGroupImpl->max_best_effort_tasks_ 和 ThreadGroupImpl->max_tasks_ 。

    当ScopedBlockingCall 析构的时候就会调用ThreadGroupImpl::WorkerThreadDelegateImpl::BlockingEnded()方法

    void ThreadGroupImpl::WorkerThreadDelegateImpl::BlockingEnded() {
  // 检查是否为WorkerThread线程内执行任务过程中调用，如果不是直接返回
  if (!read_worker().current_task_priority) {
    return;
  }

  // 清空blocking_start_time
  write_worker().blocking_start_time = TimeTicks();
  if (!incremented_max_tasks_for_shutdown_) {
    if (incremented_max_tasks_since_blocked_)
      // 增加郭max_task_ 则减少
      outer_->DecrementMaxTasksLockRequired();
    else
      // 没有增加过max_task_ 则减少未解析的阻塞任务计数
      --outer_->num_unresolved_may_block_;

    if (*read_worker().current_task_priority == TaskPriority::BEST_EFFORT) {
      if (incremented_max_best_effort_tasks_since_blocked_) 
        // 增加过max_best_effort_tasks_ 则减少
        outer_->DecrementMaxBestEffortTasksLockRequired();
      else
         //  没有增加过max_task_ 则减少未解析的阻塞任务计数
        --outer_->num_unresolved_best_effort_may_block_;
    }
  }
  // 重置incremented_max_tasks_since_blocked_ 和incremented_max_best_effort_tasks_since_blocked_
  incremented_max_tasks_since_blocked_ = false;
  incremented_max_best_effort_tasks_since_blocked_ = false;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29

    函数很简单， 更新ThreadGroupImpl->max_tasks_ 和 ThreadGroupImpl->max_best_effort_tasks_ 、ThreadGroupImpl->num_unresolved_may_block_ 和 ThreadGroupImpl->num_unresolved_best_effort_may_block_计数。

    线程优先级管理
    当前程没有达到最大并发的时候，是可以通过创建线程来解决问题的，当线程达到最大并发的时候就要考虑优先级问题：
    UpdateMinAllowedPriorityLockRequired 函数就是当线程不够的时候，用于记录下一个要执行任务的优先级。

    void ThreadGroupImpl::UpdateMinAllowedPriorityLockRequired() {
  if (priority_queue_.IsEmpty() || num_running_tasks_ < max_tasks_) {
    max_allowed_sort_key_.store(kMaxYieldSortKey, std::memory_order_relaxed);
  } else {
    max_allowed_sort_key_.store({priority_queue_.PeekSortKey().priority(),
                                 priority_queue_.PeekSortKey().worker_count()},
                                std::memory_order_relaxed);
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    当有任务运行或者任务结束的时候， 或者线程池线程数量发生变化的时候、或者有新任务添加的时候就会调用UpdateMinAllowedPriorityLockRequired函数来更新max_allowed_sort_key_的值， max_allowed_sort_key_表示当任务数量不足的时候下一个要执行的任务的优先级。当线程够用的时候max_allowed_sort_key_设置为kMaxYieldSortKey 表示优先级任务都不需要让出线程。 如果线程自觉，低于该优先级的任务就应该主动让出cpu执行，当然这是一个君子协定，幸好chromium是一个独立的产品，开发者都是内部的人，不像android操作系统。

    另外ThreadGroupImpl还提供了函数ShouldYield, 用于占用线程的任务自行判断是否要让出cpu，这就是一个君子协议

    bool ThreadPoolImpl::ShouldYield(const TaskSource* task_source) {
  const TaskPriority priority = task_source->priority_racy();
  auto* const thread_group =
      GetThreadGroupForTraits({priority, task_source->thread_policy()});
  // A task whose priority changed and is now running in the wrong thread group
  // should yield so it's rescheduled in the right one.
  if (!thread_group->IsBoundToCurrentThread())
    return true;
  return GetThreadGroupForTraits({priority, task_source->thread_policy()})
      ->ShouldYield(task_source->GetSortKey());
}

bool ThreadGroup::ShouldYield(TaskSourceSortKey sort_key) {
  DCHECK(TS_UNCHECKED_READ(max_allowed_sort_key_).is_lock_free());

  if (!task_tracker_->CanRunPriority(sort_key.priority()))
    return true;
  // It is safe to read |max_allowed_sort_key_| without a lock since this
  // variable is atomic, keeping in mind that threads may not immediately see
  // the new value when it is updated.
  auto max_allowed_sort_key =
      TS_UNCHECKED_READ(max_allowed_sort_key_).load(std::memory_order_relaxed);

  // To reduce unnecessary yielding, a task will never yield to a BEST_EFFORT
  // task regardless of its worker_count.
  if (sort_key.priority() > max_allowed_sort_key.priority ||
      max_allowed_sort_key.priority == TaskPriority::BEST_EFFORT) {
    return false;
  }
  // Otherwise, a task only yields to a task of equal priority if its
  // worker_count would be greater still after yielding, e.g. a job with 1
  // worker doesn't yield to a job with 0 workers.
  if (sort_key.priority() == max_allowed_sort_key.priority &&
      sort_key.worker_count() <= max_allowed_sort_key.worker_count + 1) {
    return false;
  }

  // Reset |max_allowed_sort_key_| so that only one thread should yield at a
  // time for a given task.
  max_allowed_sort_key =
      TS_UNCHECKED_READ(max_allowed_sort_key_)
          .exchange(kMaxYieldSortKey, std::memory_order_relaxed);
  // Another thread might have decided to yield and racily reset
  // |max_allowed_sort_key_|, in which case this thread doesn't yield.
  return max_allowed_sort_key.priority != TaskPriority::BEST_EFFORT;
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47

    到这里线程池的内部工作原理我们就分析完毕了。 下面来分析外部是如何向线程池投递任务的。

    任务投递

    根据文档我们知道chromium 支持串行任务投递和并行投递两种方式，投递串行方式的api是

    coped_refptr<SequencedTaskRunner> sequenced_task_runner =
    base::ThreadPool::CreateSequencedTaskRunner(...);

// TaskB runs after TaskA completes.
sequenced_task_runner->PostTask(FROM_HERE, base::BindOnce(&TaskA));
sequenced_task_runner->PostTask(FROM_HERE, base::BindOnce(&TaskB));

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    投递并行任务的api是

    coped_refptr<TaskRunner> task_runner = base::ThreadPool::CreateTaskRunner({base::TaskPriority::USER_VISIBLE});

// TaskB runs after TaskA completes.
task_runner->PostTask(FROM_HERE, base::BindOnce(&TaskA));
task_runner->PostTask(FROM_HERE, base::BindOnce(&TaskB));


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    我们来分析串行方式投递

    串行任务投递
    scoped_refptr<SequencedTaskRunner> ThreadPoolImpl::CreateSequencedTaskRunner(
    const TaskTraits& traits) {
  return MakeRefCounted<PooledSequencedTaskRunner>(traits, this);
}
PooledSequencedTaskRunner::PooledSequencedTaskRunner(
    const TaskTraits& traits,
    PooledTaskRunnerDelegate* pooled_task_runner_delegate)
    : pooled_task_runner_delegate_(pooled_task_runner_delegate),
      sequence_(MakeRefCounted<Sequence>(traits,
                                         this,
                                         TaskSourceExecutionMode::kSequenced)) {
}

Sequence::Sequence(const TaskTraits& traits,
                   TaskRunner* task_runner,
                   TaskSourceExecutionMode execution_mode)，
    : TaskSource(traits, task_runner, execution_mode) {}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    PooledSequencedTaskRunner->pooled_task_runner_delegate_ 是ThreadPoolImpl 实例。
    PooledSequencedTaskRunner->sequence_是sequence类的实例， 用于控制串行，我们后面会看到，并且 Sequence对象是TaskSource的子类，也就是要放到线程池的优先级任务队列上，为线程池提供任务源。

    先来分析延时任务处理

    bool PooledSequencedTaskRunner::PostDelayedTask(const Location& from_here,
                                                OnceClosure closure,
                                                TimeDelta delay) {
 ......

  Task task(from_here, std::move(closure), TimeTicks::Now(), delay,
            GetDefaultTaskLeeway());

  // Post the task as part of |sequence_|.
  return pooled_task_runner_delegate_->PostTaskWithSequence(std::move(task),
                                                            sequence_);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    投递任务调用ThreadGroupImpl->PostTaskWithSequence() 方法。 这里传递了PooledSequencedTaskRunner->sequence_ 变量

    bool ThreadPoolImpl::PostTaskWithSequence(Task task,
                                          scoped_refptr<Sequence> sequence) {
......
  if (task.delayed_run_time.is_null()) {
    // 如果没有延时，任务要求立即执行，调用PostTaskWithSequenceNow投递任务
    return PostTaskWithSequenceNow(std::move(task), std::move(sequence));
  } else {
    // It's safe to take a ref on this pointer since the caller must have a ref
    // to the TaskRunner in order to post.
    scoped_refptr<TaskRunner> task_runner = sequence->task_runner();
    // 向delayed_task_manager_ 添加一个延时任务， 任务到期后调用PostTaskWithSequenceNow投递任务。
    delayed_task_manager_.AddDelayedTask(
        std::move(task),
        BindOnce(
            [](scoped_refptr<Sequence> sequence,
               ThreadPoolImpl* thread_pool_impl, Task task) {
              thread_pool_impl->PostTaskWithSequenceNow(std::move(task),
                                                        std::move(sequence));
            },
            std::move(sequence), Unretained(this)),
        std::move(task_runner));
  }

  return true;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    函数很简单。如果是立即执行的任务就调用PostTaskWithSequenceNow 投递任务到线程池，如果是延时任务则delayed_task_manager_添加一个定时任务，定时的时间就是任务要执行的时间，当任务可以执行后调用PostTaskWithSequenceNow投递任务到线程池。 这么做是因为线程池不支持延时任务。 注意delayed_task_manager_ 在创建时候传递了一个TaskRunner， 而这个TaskRunner是使用service_thread_创建的，也就是所有的延时任务被投递到了service_thread_ 这个线程串行消耗延时时间。

    再来分析PostTaskWithSequenceNow函数

    bool ThreadPoolImpl::PostTaskWithSequenceNow(Task task,
                                             scoped_refptr<Sequence> sequence) {
  auto transaction = sequence->BeginTransaction();
  const bool sequence_should_be_queued = transaction.WillPushImmediateTask();
  RegisteredTaskSource task_source;
  if (sequence_should_be_queued) {
    // 如果有新准备好的任务（TaskSource 不在队列上，有新任务，如要添加这个TaskSource到队列上）， 把sequence这个任务包装成一个RegisteredTaskSource 对象
    task_source = task_tracker_->RegisterTaskSource(sequence);
    // We shouldn't push |task| if we're not allowed to queue |task_source|.
    if (!task_source)
      return false;
  }
  ......
  // 添加到sequence自身队列（用于控制串行）
  transaction.PushImmediateTask(std::move(task));
  if (task_source) {
    const TaskTraits traits = transaction.traits();
    // 添加sequence任务源到线程池队列
    GetThreadGroupForTraits(traits)->PushTaskSourceAndWakeUpWorkers(
        {std::move(task_source), std::move(transaction)});
  }
  return true;
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    如果这个任务是该Sequence第一个要任务，则该任务源不在线程池的任务队列上，则需要添加Sequence到线程池的任务队列。 添加任务源到线程池任务队列的方法是GetThreadGroupForTraits(traits)->PushTaskSourceAndWakeUpWorkers(
    {std::move(task_source), std::move(transaction)})。 再此之前Sequence的还会将任务添加到自身的一个队列里面，该队列用于控制任务串行执行（因为线程池本身并没有这个能力）。

    我们先来看下如何判断任务是否需要添加到线程池任务队列上, 主要依据is_immediate_变量之前的值，如果是false则表明之前Sequence的任务队列是空的，这里是添加的第一个任务，需要将Sequence 这个TaskSource添加到线程池队列。

    bool Sequence::Transaction::WillPushImmediateTask() {
  // In a Transaction.
  AnnotateLockAcquired annotate(sequence()->lock_);

  bool was_immediate =
      sequence()->is_immediate_.exchange(true, std::memory_order_relaxed);
  return !was_immediate;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    再来看下Sequence如何添加任务到自身队列的，方便我们分析Sequence 如何使任务保持串行执行。

    
void Sequence::Transaction::PushImmediateTask(Task task) {
......

  bool queue_was_empty = sequence()->queue_.empty();
  sequence()->queue_.push(std::move(task));

  if (queue_was_empty)
    sequence()->UpdateReadyTimes();

 ......
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    该函数只是简单的将任务添加到sequence->queue_ 队列中，如果队列之前是空的，那么这是第一个任务, 就调用UpdateReadyTimes() 函数来更新Sequence的可用时间（主要用于计算TaskSource优先级）。

    void Sequence::UpdateReadyTimes() {
  if (queue_.empty()) {  // 如果queue_是空的就从延时任务更新这两个时间
    latest_ready_time_.store(delayed_queue_.top().latest_delayed_run_time(),
                             std::memory_order_relaxed);
    earliest_ready_time_.store(delayed_queue_.top().earliest_delayed_run_time(),
                               std::memory_order_relaxed);
    return;
  }

  if (delayed_queue_.empty()) { // 延时任务队列为空， latest_ready_time_ 就是第一个立即执行任务的入队列时间
    latest_ready_time_.store(queue_.front().queue_time,
                             std::memory_order_relaxed);
  } else {
    // 立即执行任务队列和延时任务队列都不为空，那么latest_ready_time_为 第一个立即执行任务的入队列时间 和 第一个延时任务的latest_delayed_run_time() 比较早的哪一个
    latest_ready_time_.store(
        std::min(queue_.front().queue_time,
                 delayed_queue_.top().latest_delayed_run_time()),
        std::memory_order_relaxed);
  }
  // 有立即执行任务，自然earliest_ready_time_ 就是0，需要马上执行
  earliest_ready_time_.store(TimeTicks(), std::memory_order_relaxed);
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    延时任务有一个运行时间范围，包括最晚运行时间和最早运行时间，延时任务一般应该在这个时间范围内运行，UpdateReadyTimes函数就是用于更新整个sequence的最早运行时间和最晚运行时间的。 另外Sequence有两个队列，一个用于存放立即执行的任务，队列变量名称为queue_， 另一个用于存放延时任务，名字叫delayed_queue_， 实际上目前chromium并没有使用delayed_queue_， 我们前面也看到了，延时任务的延时是在ThreadPoolImpl的delayed_task_manager_中消耗掉的。
    latest_ready_time_ ： 表示Sequence 这个任务源最晚需要运行的时间
    earliest_ready_time_：表示Sequence 这个任务源最早需要运行的时间
    函数的具体逻辑比较简单，请读者自行分析。

    再来看一下Sequence 被添加到线程池任务源队列的情景

    ThreadGroup* ThreadPoolImpl::GetThreadGroupForTraits(const TaskTraits& traits) {
  if (traits.priority() == TaskPriority::BEST_EFFORT &&
      traits.thread_policy() == ThreadPolicy::PREFER_BACKGROUND &&
      background_thread_group_) {
    return background_thread_group_.get();
  }

  if (traits.priority() <= TaskPriority::USER_VISIBLE &&
      traits.thread_policy() == ThreadPolicy::PREFER_BACKGROUND &&
      utility_thread_group_) {
    return utility_thread_group_.get();
  }

  return foreground_thread_group_.get();
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    GetThreadGroupForTraits 函数根据TaskTraits 选择一个线程池组。

    RegisteredTaskSourceAndTransaction::RegisteredTaskSourceAndTransaction(
    RegisteredTaskSource task_source_in,
    TaskSource::Transaction transaction_in)
    : task_source(std::move(task_source_in)),
      transaction(std::move(transaction_in)) {
  DCHECK_EQ(task_source.get(), transaction.task_source());
}


void ThreadGroupImpl::PushTaskSourceAndWakeUpWorkers(
    RegisteredTaskSourceAndTransaction transaction_with_task_source) {
  ScopedCommandsExecutor executor(this);
  PushTaskSourceAndWakeUpWorkersImpl(&executor,
                                     std::move(transaction_with_task_source));
}

void ThreadGroup::PushTaskSourceAndWakeUpWorkersImpl(
    BaseScopedCommandsExecutor* executor,
    RegisteredTaskSourceAndTransaction transaction_with_task_source) {
  CheckedAutoLock auto_lock(lock_);
......
  // 获取排序的key
  auto sort_key = transaction_with_task_source.task_source->GetSortKey();
......
  transaction_with_task_source.transaction.Release();
  // 添加到优先级队列
  priority_queue_.Push(std::move(transaction_with_task_source.task_source),
                       sort_key);
  // 尝试补充或者唤醒线程
  EnsureEnoughWorkersLockRequired(executor);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31

    这里参数RegisteredTaskSourceAndTransaction 就是持有一个TaskSource（这里是Sequence） 和TaskSource::Transaction的对象。
    PushTaskSourceAndWakeUpWorkers 函数主要计算排序用的key，然后添加到线程组的队列里面（会根据排序key排序）。

    TaskSourceSortKey Sequence::GetSortKey() const {
  return TaskSourceSortKey(
      priority_racy(),
      TS_UNCHECKED_READ(latest_ready_time_).load(std::memory_order_relaxed));
}
bool TaskSourceSortKey::operator<(const TaskSourceSortKey& other) const {
  // This TaskSourceSortKey is considered more important than |other| if it has
  // a higher priority or if it has the same priority but fewer workers, or if
  // it has the same priority and same worker count but its next task was
  // posted sooner than |other|'s.

  // A lower priority is considered less important.
  if (priority_ != other.priority_)
    return priority_ < other.priority_;

  // A greater worker count is considered less important.
  if (worker_count_ != other.worker_count_)
    return worker_count_ > other.worker_count_;

  // Lastly, a greater ready time is considered less important.
  return ready_time_ > other.ready_time_;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    排序key主要根据优先级，最晚执行时间进行计算。还会参考服务在这个TaskSource的 线程运量 。 任务源进队列之后就是任务执行的过程了。我们前面已经分析了。

    当线程池要为TaskSource服务时会调用TaskSource->TaskTask() 获取Task

    Task Sequence::TakeTask(TaskSource::Transaction* transaction) {
 ......
  auto next_task = TakeEarliestTask();

  if (!IsEmpty())
    UpdateReadyTimes();

  return next_task;
}
Task Sequence::TakeEarliestTask() {
  if (queue_.empty())
    return delayed_queue_.take_top();

  if (delayed_queue_.empty())
    return TakeNextImmediateTask();

  // Both queues contain at least a task. Decide from which one the task should
  // be taken.
  if (queue_.front().queue_time <=
      delayed_queue_.top().latest_delayed_run_time())
    return TakeNextImmediateTask();

  return delayed_queue_.take_top();
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    该方法会从queue_ 或者 delayed_queue_ 上获取任务返回， 选择任务的原则是找最着急运行的任务执行。

    当执行完一个TaskSource上的任务后，还会调用TaskSource上是否还有其他任务需要执行， 并且给TaskSource机会做一些清理工作

    bool Sequence::DidProcessTask(TaskSource::Transaction* transaction) {
......
  // See comment on TaskSource::task_runner_ for lifetime management details.
  if (IsEmpty()) {
    is_immediate_.store(false, std::memory_order_relaxed);
    ReleaseTaskRunner();
    return false;
  }

  // Let the caller re-enqueue this non-empty Sequence regardless of
  // |run_result| so it can continue churning through this Sequence's tasks and
  // skip/delete them in the proper scope.
  return true;
}
bool Sequence::IsEmpty() const {
  return queue_.empty() && delayed_queue_.empty();
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    DidProcessTask 函数判断如果任务队列是空的，则设置is_immediate_=false， 这样有新任务到来就会将TaskSource重新添加到线程池的线程组队列。 我们也会看到在一个WorkerThread 一轮循环过程中, Sequeue 是不会被重新添加到线程组的队列里面的（is_immediate_ 这个值控制）， 所以可以做到一个Sequeue 上的任务是串行执行的。但是Sequeue 不保证执行的WorkerThread， 并且Sequeue 的锁可以保证在同一个Sequeue 上执行任务结果的可见性。

    到这里串行执行任务我们就分析完了。

    并行任务投递

    前面我们知道串行执行靠Sequeue 这个特殊的TaskSource控制串行，那我们来分析下并行投递是如何做到的

    scoped_refptr<TaskRunner> ThreadPoolImpl::CreateTaskRunner(
    const TaskTraits& traits) {
  return MakeRefCounted<PooledParallelTaskRunner>(traits, this);
}


    • 1
    • 2
    • 3
    • 4
    • 5

    并行投递使用的是PooledParallelTaskRunner作为TaskRunner

    bool PooledParallelTaskRunner::PostDelayedTask(const Location& from_here,
                                               OnceClosure closure,
                                               TimeDelta delay) {
  if (!PooledTaskRunnerDelegate::MatchesCurrentDelegate(
          pooled_task_runner_delegate_)) {
    return false;
  }

  // Post the task as part of a one-off single-task Sequence.
  scoped_refptr<Sequence> sequence = MakeRefCounted<Sequence>(
      traits_, this, TaskSourceExecutionMode::kParallel);

  return pooled_task_runner_delegate_->PostTaskWithSequence(
      Task(from_here, std::move(closure), TimeTicks::Now(), delay),
      std::move(sequence));
}

bool TaskRunner::PostTask(const Location& from_here, OnceClosure task) {
  return PostDelayedTask(from_here, std::move(task), base::TimeDelta());
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21

    我们可以看到每次投递任务，都会创建一个新的Sequence， 这样就能保证任务并行执行了。

    总结

    到这里，除了关闭的情景，线程池的方方面面也分析的差不多了。 我觉得chromium的设计实在是太复杂，存在过度设计嫌疑。 另外如果有协程就不必这么大费周折的去处理阻塞任务。 也许是因为开发者对底层不太了解，也可能是很多历史包袱，才使得chromium的设计如此复杂吧。

  • 相关阅读:
    .NET下数据库的负载均衡(有趣实验)
    Nginx-高可用及 KeepAlived 配置
    SpringCloud Bus消息总线
    深度优先搜索算法-dfs讲解
    微服务技术栈-Nacos配置管理和Feign远程调用
    中小企业如何差异化“生意表达”，成为最了不起的小企业？
    【PyTorch】3-基础实战（ResNet）
    8.springboot的自定义配置（@value注解和@ConfigurationProperties）
    网上那么多教人赚钱的方法，但是你实际上是靠什么赚钱的呢？
    .net6部署到linux上（CentOS Linux 7）
  • 原文地址：https://blog.csdn.net/woai110120130/article/details/133694109