post_job.h 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212
  1. // Copyright 2019 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #ifndef BASE_TASK_POST_JOB_H_
  5. #define BASE_TASK_POST_JOB_H_
  6. #include <limits>
  7. #include "base/base_export.h"
  8. #include "base/callback.h"
  9. #include "base/dcheck_is_on.h"
  10. #include "base/location.h"
  11. #include "base/memory/raw_ptr.h"
  12. namespace base {
  13. namespace internal {
  14. class JobTaskSource;
  15. class PooledTaskRunnerDelegate;
  16. }
  17. class TaskTraits;
  18. enum class TaskPriority : uint8_t;
  19. // Delegate that's passed to Job's worker task, providing an entry point to
  20. // communicate with the scheduler. To prevent deadlocks, JobDelegate methods
  21. // should never be called while holding a user lock.
  22. class BASE_EXPORT JobDelegate {
  23. public:
  24. // A JobDelegate is instantiated for each worker task that is run.
  25. // |task_source| is the task source whose worker task is running with this
  26. // delegate and |pooled_task_runner_delegate| is used by ShouldYield() to
  27. // check whether the pool wants this worker task to yield (null if this worker
  28. // should never yield -- e.g. when the main thread is a worker).
  29. JobDelegate(internal::JobTaskSource* task_source,
  30. internal::PooledTaskRunnerDelegate* pooled_task_runner_delegate);
  31. JobDelegate(const JobDelegate&) = delete;
  32. JobDelegate& operator=(const JobDelegate&) = delete;
  33. ~JobDelegate();
  34. // Returns true if this thread *must* return from the worker task on the
  35. // current thread ASAP. Workers should periodically invoke ShouldYield (or
  36. // YieldIfNeeded()) as often as is reasonable.
  37. bool ShouldYield();
  38. // If ShouldYield(), this will pause the current thread (allowing it to be
  39. // replaced in the pool); no-ops otherwise. If it pauses, it will resume and
  40. // return from this call whenever higher priority work completes.
  41. // Prefer ShouldYield() over this (only use YieldIfNeeded() when unwinding
  42. // the stack is not possible).
  43. void YieldIfNeeded();
  44. // Notifies the scheduler that max concurrency was increased, and the number
  45. // of worker should be adjusted accordingly. See PostJob() for more details.
  46. void NotifyConcurrencyIncrease();
  47. // Returns a task_id unique among threads currently running this job, such
  48. // that GetTaskId() < worker count. To achieve this, the same task_id may be
  49. // reused by a different thread after a worker_task returns.
  50. uint8_t GetTaskId();
  51. // Returns true if the current task is called from the thread currently
  52. // running JobHandle::Join().
  53. bool IsJoiningThread() const {
  54. return pooled_task_runner_delegate_ == nullptr;
  55. }
  56. private:
  57. static constexpr uint8_t kInvalidTaskId = std::numeric_limits<uint8_t>::max();
  58. const raw_ptr<internal::JobTaskSource> task_source_;
  59. const raw_ptr<internal::PooledTaskRunnerDelegate>
  60. pooled_task_runner_delegate_;
  61. uint8_t task_id_ = kInvalidTaskId;
  62. #if DCHECK_IS_ON()
  63. // Value returned by the last call to ShouldYield().
  64. bool last_should_yield_ = false;
  65. #endif
  66. };
  67. // Handle returned when posting a Job. Provides methods to control execution of
  68. // the posted Job. To prevent deadlocks, JobHandle methods should never be
  69. // called while holding a user lock.
  70. class BASE_EXPORT JobHandle {
  71. public:
  72. JobHandle();
  73. JobHandle(const JobHandle&) = delete;
  74. JobHandle& operator=(const JobHandle&) = delete;
  75. // A job must either be joined, canceled or detached before the JobHandle is
  76. // destroyed.
  77. ~JobHandle();
  78. JobHandle(JobHandle&&);
  79. JobHandle& operator=(JobHandle&&);
  80. // Returns true if associated with a Job.
  81. explicit operator bool() const { return task_source_ != nullptr; }
  82. // Returns true if there's any work pending or any worker running.
  83. bool IsActive() const;
  84. // Update this Job's priority.
  85. void UpdatePriority(TaskPriority new_priority);
  86. // Notifies the scheduler that max concurrency was increased, and the number
  87. // of workers should be adjusted accordingly. See PostJob() for more details.
  88. void NotifyConcurrencyIncrease();
  89. // Contributes to the job on this thread. Doesn't return until all tasks have
  90. // completed and max concurrency becomes 0. This also promotes this Job's
  91. // priority to be at least as high as the calling thread's priority. When
  92. // called immediately, prefer CreateJob(...).Join() over PostJob(...).Join()
  93. // to avoid having too many workers scheduled for executing the workload.
  94. void Join();
  95. // Forces all existing workers to yield ASAP. Waits until they have all
  96. // returned from the Job's callback before returning.
  97. void Cancel();
  98. // Forces all existing workers to yield ASAP but doesn’t wait for them.
  99. // Warning, this is dangerous if the Job's callback is bound to or has access
  100. // to state which may be deleted after this call.
  101. void CancelAndDetach();
  102. // Can be invoked before ~JobHandle() to avoid waiting on the job completing.
  103. void Detach();
  104. private:
  105. friend class internal::JobTaskSource;
  106. explicit JobHandle(scoped_refptr<internal::JobTaskSource> task_source);
  107. scoped_refptr<internal::JobTaskSource> task_source_;
  108. };
  109. // Callback used in PostJob() to control the maximum number of threads calling
  110. // the worker task concurrently.
  111. // Returns the maximum number of threads which may call a job's worker task
  112. // concurrently. |worker_count| is the number of threads currently assigned to
  113. // this job which some callers may need to determine their return value.
  114. using MaxConcurrencyCallback =
  115. RepeatingCallback<size_t(size_t /*worker_count*/)>;
  116. // Posts a repeating |worker_task| with specific |traits| to run in parallel on
  117. // base::ThreadPool.
  118. // Returns a JobHandle associated with the Job, which can be joined, canceled or
  119. // detached.
  120. // ThreadPool APIs, including PostJob() and methods of the returned JobHandle,
  121. // must never be called while holding a lock that could be acquired by
  122. // |worker_task| or |max_concurrency_callback| -- that could result in a
  123. // deadlock. This is because [1] |max_concurrency_callback| may be invoked while
  124. // holding internal ThreadPool lock (A), hence |max_concurrency_callback| can
  125. // only use a lock (B) if that lock is *never* held while calling back into a
  126. // ThreadPool entry point from any thread (A=>B/B=>A deadlock) and [2]
  127. // |worker_task| or |max_concurrency_callback| is invoked synchronously from
  128. // JobHandle::Join() (A=>JobHandle::Join()=>A deadlock).
  129. // To avoid scheduling overhead, |worker_task| should do as much work as
  130. // possible in a loop when invoked, and JobDelegate::ShouldYield() should be
  131. // periodically invoked to conditionally exit and let the scheduler prioritize
  132. // work.
  133. //
  134. // A canonical implementation of |worker_task| looks like:
  135. // void WorkerTask(JobDelegate* job_delegate) {
  136. // while (!job_delegate->ShouldYield()) {
  137. // auto work_item = worker_queue.TakeWorkItem(); // Smallest unit of work.
  138. // if (!work_item)
  139. // return:
  140. // ProcessWork(work_item);
  141. // }
  142. // }
  143. //
  144. // |max_concurrency_callback| controls the maximum number of threads calling
  145. // |worker_task| concurrently. |worker_task| is only invoked if the number of
  146. // threads previously running |worker_task| was less than the value returned by
  147. // |max_concurrency_callback|. In general, |max_concurrency_callback| should
  148. // return the latest number of incomplete work items (smallest unit of work)
  149. // left to processed. JobHandle/JobDelegate::NotifyConcurrencyIncrease() *must*
  150. // be invoked shortly after |max_concurrency_callback| starts returning a value
  151. // larger than previously returned values. This usually happens when new work
  152. // items are added and the API user wants additional threads to invoke
  153. // |worker_task| concurrently. The callbacks may be called concurrently on any
  154. // thread until the job is complete. If the job handle is detached, the
  155. // callbacks may still be called, so they must not access global state that
  156. // could be destroyed.
  157. //
  158. // |traits| requirements:
  159. // - base::ThreadPolicy must be specified if the priority of the task runner
  160. // will ever be increased from BEST_EFFORT.
  161. JobHandle BASE_EXPORT PostJob(const Location& from_here,
  162. const TaskTraits& traits,
  163. RepeatingCallback<void(JobDelegate*)> worker_task,
  164. MaxConcurrencyCallback max_concurrency_callback);
  165. // Creates and returns a JobHandle associated with a Job. Unlike PostJob(), this
  166. // doesn't immediately schedules |worker_task| to run on base::ThreadPool
  167. // workers; the Job is then scheduled by calling either
  168. // NotifyConcurrencyIncrease() or Join().
  169. JobHandle BASE_EXPORT
  170. CreateJob(const Location& from_here,
  171. const TaskTraits& traits,
  172. RepeatingCallback<void(JobDelegate*)> worker_task,
  173. MaxConcurrencyCallback max_concurrency_callback);
  174. } // namespace base
  175. #endif // BASE_TASK_POST_JOB_H_