post_job.h 7.3 KB

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