【Chromium】Base库的ThreadPool

struct BASE_EXPORT InitParams{
	enum class CommonThreadPoolEnvironment {
      // 使用默认环境（无环境）。
      DEFAULT,
      #if defined(OS_WIN)
      // 将线程池的工作线程放置在 COM MTA 中。
      COM_MTA,
      // 将线程池的前台工作线程放置在 COM STA 中。这是为了模拟SequencedWorkerPool和BrowserThreadImpl的行为，而ThreadPool已经取代了它们。需要 COM STA 的任务应该使用CreateCOMSTATaskRunner()而不是Create(Sequenced)TaskRunnerWithTraits() + 这个初始化参数。
      DEPRECATED_COM_STA_IN_FOREGROUND_GROUP,
      #endif  // defined(OS_WIN)
    };

    InitParams(int max_num_foreground_threads_in);
    ~InitParams();

    // 前台线程组中可以同时运行的未阻塞任务的最大数量。
    int max_num_foreground_threads;

    // 在运行序列化和并行任务时，是否初始化 COM。
    CommonThreadPoolEnvironment common_thread_pool_environment = CommonThreadPoolEnvironment::DEFAULT;

    // 建议的未使用线程可以回收的时间间隔。
    TimeDelta suggested_reclaim_time = TimeDelta::FromSeconds(30);
};

// 用于观察 ThreadPool 工作线程的主函数入口和退出的接口。

class WorkerThreadObserver {
public:
    virtual ~WorkerThreadObserver() = default;

    // 在 ThreadPool 工作线程的主函数开始之前调用，任何任务运行之前。
    virtual void OnWorkerThreadMainEntry() = 0;

    // 在 ThreadPool 工作线程的主函数结束时调用，无法再运行任务时。
    virtual void OnWorkerThreadMainExit() = 0;
};

// ScopedExecutionFence（最佳尝试）防止在其作用域内在ThreadPoolInstance中调度任何/BEST_EFFORT优先级的新任务。在其销毁时，被抢占的任务将被释放。
// 注意：ScopedExecutionFence的构造函数不会等待当前正在运行的任务（因为它们在进入此作用域之前发布，并且不违反契约；其中一些可能是CONTINUE_ON_SHUTDOWN，并且等待它们完成是不明智的）。

class BASE_EXPORT ScopedExecutionFence {
public:
    ScopedExecutionFence();
    ~ScopedExecutionFence();

private:
    DISALLOW_COPY_AND_ASSIGN(ScopedExecutionFence);
};

class BASE_EXPORT ScopedBestEffortExecutionFence {
public:
    ScopedBestEffortExecutionFence();
    ~ScopedBestEffortExecutionFence();

private:
    DISALLOW_COPY_AND_ASSIGN(ScopedBestEffortExecutionFence);
};

// 在线上环境中不允许销毁 ThreadPoolInstance，它将始终被泄漏。
// 在测试中，只有在 JoinForTesting() 返回之后才应该销毁 ThreadPoolInstance。
virtual ~ThreadPoolInstance() = default;

// 创建一个准备就绪的线程池。参数 |name| 用于标记，它不能为空。它应该标识创建 ThreadPoolInstance 的组件。
// 线程池在调用 Start() 之前不会创建线程。任务可以随时发布，但直到调用 Start() 后才会运行。在测试中，建议使用 base::test::ScopedTaskEnvironment（确保隔离性）。

static void Create(StringPiece name);

// 使用默认参数创建并启动一个线程池。参数 |name| 用于标记，它不能为空。它应该标识调用此方法的组件。
// 此方法会调用 Start()，之后再次调用它是无效的。如果失败，会触发 CHECK。
// 在测试中，建议使用 base::test::ScopedTaskEnvironment（确保隔离性）。
static void CreateAndStartWithDefaultParams(StringPiece name);

// 允许线程池根据 |init_params| 规范创建线程并运行任务。 
// // 如果指定了 |worker_thread_observer|，当一个工作线程进入和退出其主函数时，它将被通知。
// // 在 JoinForTesting() 返回之前，它不能被销毁（在生产环境中不能被销毁）。 
// // 如果失败，会触发 CHECK。 
virtual void Start( const InitParams& init_params, WorkerThreadObserver* worker_thread_observer = nullptr) = 0;

// 与 CreateAndStartWithDefaultParams() 相同，但允许调用者将 Create() 和 StartWithDefaultParams() 分开调用。
void StartWithDefaultParams();

// 等待直到没有待处理的非延迟任务。可以在测试中调用此方法，在所有非延迟任务运行后验证条件是否满足。
// 不等待延迟任务。在调用期间等待从其他线程发布的非延迟任务。当关闭完成时立即返回。
virtual void FlushForTesting() = 0;

// 当没有未完成的非延迟任务时，返回并调用 |flush_callback|。|flush_callback| 可以在任何线程上被调用，不应执行大量的工作。
// 当需要在刷新期间执行当前线程上的其他工作时，可以使用此方法。一次只能有一个 FlushAsyncForTesting() 在等待。
virtual void FlushAsyncForTesting(OnceClosure flush_callback) = 0;

// 加入所有线程。已经运行的任务被允许完成执行。此方法只能调用一次。在此调用期间或之后，不允许使用此线程池实例创建任务运行器或发布任务。
virtual void JoinForTesting() = 0;

// Shutdown() 是一个同步关闭线程池的方法。一旦调用此方法，只有使用 BLOCK_SHUTDOWN 行为发布的任务将会被运行。当该方法返回时：

// 所有已经运行的 SKIP_ON_SHUTDOWN 任务已经完成执行。
// 所有已发布的 BLOCK_SHUTDOWN 任务已经完成执行。CONTINUE_ON_SHUTDOWN 任务可能仍在运行。
// 需要注意的是，实现可以保持线程和其他资源的活动状态，以便在 Shutdown() 返回后继续运行 CONTINUE_ON_SHUTDOWN 任务。此方法只能调用一次。
virtual void Shutdown() = 0;

// 将 |thread_pool| 注册为处理通过 post_task.h API 发布的任务的线程池。在测试中，推荐使用 base::test::ScopedTaskEnvironment（确保隔离）。
static void Set(std::unique_ptr<ThreadPoolInstance> thread_pool);


// 获取通过 SetInstance() 或 Create() 设置的 ThreadPoolInstance。
// 应该非常少使用此方法；大多数线程池的使用者应该使用 post_task.h API。
// 特别是，不要这样做： if (!ThreadPoolInstance::Get()) { ThreadPoolInstance::Set(...); base::PostTask(...); } 
// 而是确保在进程初始化阶段的一个确定的位置尽早调用 SetInstance()。
static ThreadPoolInstance* Get();

  // Sets whether a fence prevents execution of tasks of any / BEST_EFFORT
  // priority.
  virtual void SetHasFence(bool can_run) = 0;
  virtual void SetHasBestEffortFence(bool can_run) = 0;


// 返回在此线程池中可以同时运行的具有 |traits| 的非单线程非阻塞任务的最大数量。|traits| 不能包含 TaskPriority::BEST_EFFORT。// 请不要使用此方法。
// 要处理 n 个项，请发布 n 个任务，每个任务处理一个项，而不是使用 GetMaxConcurrentNonBlockedTasksWithTraitsDeprecated() 任务，每个任务处理 n/GetMaxConcurrentNonBlockedTasksWithTraitsDeprecated() 个项。
int GetMaxConcurrentNonBlockedTasksWithTraitsDeprecated( const TaskTraits& traits) const = 0;

class BASE_EXPORT ThreadPoolImpl : public ThreadPoolInstance,
                                   public TaskExecutor,
                                   public ThreadGroup::Delegate,
                                   public PooledTaskRunnerDelegate

// 使用生产 TaskTracker 创建 ThreadPoolImpl。 
// |histogram_label| 用于标记，不能为空。 
explicit ThreadPoolImpl(StringPiece histogram_label);
// 仅用于测试。使用自定义 TaskTracker 和 TickClock 创建 ThreadPoolImpl。 
ThreadPoolImpl(StringPiece histogram_label, 
               std::unique_ptr<TaskTrackerImpl> task_tracker,
               const TickClock* tick_clock);
~ThreadPoolImpl() override;

  // ThreadPoolInstance:
  void Start(const ThreadPoolInstance::InitParams& init_params,
             WorkerThreadObserver* worker_thread_observer) override;
  int GetMaxConcurrentNonBlockedTasksWithTraitsDeprecated(
      const TaskTraits& traits) const override;
  void Shutdown() override;
  void FlushForTesting() override;
  void FlushAsyncForTesting(OnceClosure flush_callback) override;
  void JoinForTesting() override;
  void SetHasFence(bool has_fence) override;
  void SetHasBestEffortFence(bool has_best_effort_fence) override;

  // TaskExecutor:
  bool PostDelayedTask(const Location& from_here,
                       const TaskTraits& traits,
                       OnceClosure task,
                       TimeDelta delay) override;
  scoped_refptr<TaskRunner> CreateTaskRunner(const TaskTraits& traits) override;
  scoped_refptr<SequencedTaskRunner> CreateSequencedTaskRunner(const TaskTraits& traits) override;
  scoped_refptr<SingleThreadTaskRunner> CreateSingleThreadTaskRunner(const TaskTraits& traits,SingleThreadTaskRunnerThreadMode thread_mode) override;
#if defined(OS_WIN)
  scoped_refptr<SingleThreadTaskRunner> CreateCOMSTATaskRunner(const TaskTraits& traits,SingleThreadTaskRunnerThreadMode thread_mode) override;
#endif  // defined(OS_WIN)
  scoped_refptr<UpdateableSequencedTaskRunner> CreateUpdateableSequencedTaskRunner(const TaskTraits& traits);

// 返回 ThreadPool 上调度的下一个任务的 TimeTicks（如果是立即执行，则返回 Now()；如果没有任务，则返回 nullopt）。
// 这是线程安全的，即使在调用此方法时并行发布任务，也是安全的，但这种情况下可能存在竞争，无法确定此调用是否及时看到新任务。
Optional<TimeTicks> NextScheduledRunTimeForTesting() const;

// 强制发布成熟的延迟任务（例如，当时间被模拟并且比 ServiceThread 上的实时延迟更快时）。
void ProcessRipeDelayedTasksForTesting();

// 在更新 |has_fence_| 或 |has_best_effort_fence_| 后调用。设置 TaskTracker 中的 CanRunPolicy，并根据需要唤醒工作线程。
void UpdateCanRunPolicy();

// 如果设置了 |all_tasks_user_blocking_|，则将 |traits| 的优先级设置为 TaskPriority::USER_BLOCKING。
TaskTraits SetUserBlockingPriorityIfNeeded(TaskTraits traits) const;

// 报告心跳指标。
void ReportHeartbeatMetrics() const;

// 获取适用于给定 TaskTraits 的 ThreadGroup。
const ThreadGroup* GetThreadGroupForTraits(const TaskTraits& traits) const;

// ThreadGroup::Delegate:
// 获取适用于给定 TaskTraits 的 ThreadGroup。
ThreadGroup* GetThreadGroupForTraits(const TaskTraits& traits) override;

// 将 |task| 作为 |sequence| 的一部分发布以由适当的线程组执行。在 |task| 经过 TaskTracker::WillPostTask() 和 |task| 的延迟运行时间之后才能调用此方法。
bool PostTaskWithSequenceNow(Task task, scoped_refptr<Sequence> sequence);

// PooledTaskRunnerDelegate:
bool PostTaskWithSequence(Task task, scoped_refptr<Sequence> sequence) override;
bool EnqueueJobTaskSource(scoped_refptr<JobTaskSource> task_source) override;
bool IsRunningPoolWithTraits(const TaskTraits& traits) const override;
void UpdatePriority(scoped_refptr<TaskSource> task_source, TaskPriority priority) override;

// base/task/thread_pool 中所有类使用的时钟实例。必须比其他所有内容存活时间更长，以确保 
Now() 的一致性。
ThreadPoolClock thread_pool_clock_;

const std::unique_ptr<TaskTrackerImpl> task_tracker_;
std::unique_ptr<Thread> service_thread_;
DelayedTaskManager delayed_task_manager_;
PooledSingleThreadTaskRunnerManager single_thread_task_runner_manager_;

// 表示所有任务都被视为以 TaskPriority::USER_BLOCKING 发布。由于此标志在 Start() 中设置，因此不适用于在 Start() 之前发布的任务或在 Start() 之前创建的 TaskRunner。
// TODO(fdoray): 在实验后删除。https://crbug.com/757022
AtomicFlag all_tasks_user_blocking_;

std::unique_ptr<ThreadGroup> foreground_thread_group_;
std::unique_ptr<ThreadGroupImpl> background_thread_group_;

// 指示此 TaskScheduler 是否已启动。访问受 |sequence_checker_| 控制。
bool started_ = false;

// 是否 --disable-best-effort-tasks 开关阻止执行 BEST_EFFORT 任务直到关闭。
const bool has_disable_best_effort_switch_;

// 是否有一个阻止执行任何/ BEST_EFFORT 优先级任务的 fence。访问受 |sequence_checker_| 控制。
bool has_fence_ = false;
bool has_best_effort_fence_ = false;

#if DCHECK_IS_ON()
// 一旦 JoinForTesting() 返回，就会设置。
AtomicFlag join_for_testing_returned_;
#endif

#if defined(OS_WIN) && defined(COM_INIT_CHECK_HOOK_ENABLED)
// 为支持的构建提供 COM 初始化验证。
base::win::ComInitCheckHook com_init_check_hook_;
#endif

// 断言操作与 Start() 顺序一致。
SEQUENCE_CHECKER(sequence_checker_);

TrackedRefFactoryThreadGroup::Delegate tracked_ref_factory_;

// 不允许拷贝
DISALLOW_COPY_AND_ASSIGN(ThreadPoolImpl);

#define kMaxNumForegroundThreads 4;

class MyObserver : public base::WorkerThreadObserver {
public:
    // Invoked at the beginning of the main function of a ThreadPool worker,
    // before any task runs.
    void OnWorkerThreadMainEntry() override {
        L_TRACE(L"%s", __FUNCTIONW__);
    }

    // Invoked at the end of the main function of a ThreadPool worker, when it
    // can no longer run tasks.
    void OnWorkerThreadMainExit() override {
        L_TRACE(L"%s", __FUNCTIONW__);
    };
};


int main() {
    // 定义最大线程池后台线程个数
    int max_num_foreground_threads = kMaxNumForegroundThreads;
    // 初始化线程池参数
    base::ThreadPoolInstance::InitParams init_params(max_num_foreground_threads);

  	// 创建线程池实例，此时线程未启动
    std::unique_ptr<base::internal::ThreadPoolImpl> thread_pool_ = std::make_unique<base::internal::ThreadPoolImpl>("Test Thread Pool");

    // 定义线程池工作线程观察者
    MyObserver observer;

    // 线程池线程启动，不会一下子到最大线程数
    thread_pool_->Start(init_params, &observer);
  
    // 定义线程任务细节
    base::TaskTraits traits;
    for (int i = 0; i < 10; ++i) {
        // 抛延时异步任务
        thread_pool_->PostDelayedTask(FROM_HERE, traits, base::BindOnce(static_cast<void (*)(base::TimeDelta)>(
            &base::PlatformThread::Sleep),
            base::TimeDelta::FromMilliseconds(20)), base::TimeDelta::FromMilliseconds(20));
    }
    L_TRACE(L"%s", __FUNCTIONW__);
    
    // 等待线程池所有线程任务做完
    thread_pool_->FlushForTesting();
    // 等待线程池所有线程退出
    thread_pool_->JoinForTesting();

    L_TRACE(L"%s", __FUNCTIONW__);
}