post_job.h 7.9 KB

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