post_job.cc 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192
  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. #include "base/task/post_job.h"
  5. #include "base/task/scoped_set_task_priority_for_current_thread.h"
  6. #include "base/task/thread_pool/job_task_source.h"
  7. #include "base/task/thread_pool/pooled_task_runner_delegate.h"
  8. #include "base/task/thread_pool/thread_pool_impl.h"
  9. #include "base/task/thread_pool/thread_pool_instance.h"
  10. namespace base {
  11. namespace {
  12. scoped_refptr<internal::JobTaskSource> CreateJobTaskSource(
  13. const Location& from_here,
  14. const TaskTraits& traits,
  15. RepeatingCallback<void(JobDelegate*)> worker_task,
  16. MaxConcurrencyCallback max_concurrency_callback) {
  17. DCHECK(ThreadPoolInstance::Get())
  18. << "Hint: if this is in a unit test, you're likely merely missing a "
  19. "base::test::TaskEnvironment member in your fixture.\n";
  20. // ThreadPool is implicitly the destination for PostJob(). Extension traits
  21. // cannot be used.
  22. DCHECK_EQ(traits.extension_id(),
  23. TaskTraitsExtensionStorage::kInvalidExtensionId);
  24. return base::MakeRefCounted<internal::JobTaskSource>(
  25. from_here, traits, std::move(worker_task),
  26. std::move(max_concurrency_callback),
  27. static_cast<internal::ThreadPoolImpl*>(ThreadPoolInstance::Get()));
  28. }
  29. } // namespace
  30. JobDelegate::JobDelegate(
  31. internal::JobTaskSource* task_source,
  32. internal::PooledTaskRunnerDelegate* pooled_task_runner_delegate)
  33. : task_source_(task_source),
  34. pooled_task_runner_delegate_(pooled_task_runner_delegate) {
  35. DCHECK(task_source_);
  36. }
  37. JobDelegate::~JobDelegate() {
  38. if (task_id_ != kInvalidTaskId)
  39. task_source_->ReleaseTaskId(task_id_);
  40. }
  41. bool JobDelegate::ShouldYield() {
  42. #if DCHECK_IS_ON()
  43. // ShouldYield() shouldn't be called again after returning true.
  44. DCHECK(!last_should_yield_);
  45. #endif // DCHECK_IS_ON()
  46. const bool should_yield =
  47. task_source_->ShouldYield() ||
  48. (pooled_task_runner_delegate_ &&
  49. pooled_task_runner_delegate_->ShouldYield(task_source_));
  50. #if DCHECK_IS_ON()
  51. last_should_yield_ = should_yield;
  52. #endif // DCHECK_IS_ON()
  53. return should_yield;
  54. }
  55. void JobDelegate::YieldIfNeeded() {
  56. // TODO(crbug.com/839091): Implement this.
  57. }
  58. void JobDelegate::NotifyConcurrencyIncrease() {
  59. task_source_->NotifyConcurrencyIncrease();
  60. }
  61. uint8_t JobDelegate::GetTaskId() {
  62. if (task_id_ == kInvalidTaskId)
  63. task_id_ = task_source_->AcquireTaskId();
  64. return task_id_;
  65. }
  66. JobHandle::JobHandle() = default;
  67. JobHandle::JobHandle(scoped_refptr<internal::JobTaskSource> task_source)
  68. : task_source_(std::move(task_source)) {}
  69. JobHandle::~JobHandle() {
  70. DCHECK(!task_source_)
  71. << "The Job must be cancelled, detached or joined before its "
  72. "JobHandle is destroyed.";
  73. }
  74. JobHandle::JobHandle(JobHandle&&) = default;
  75. JobHandle& JobHandle::operator=(JobHandle&& other) {
  76. DCHECK(!task_source_)
  77. << "The Job must be cancelled, detached or joined before its "
  78. "JobHandle is re-assigned.";
  79. task_source_ = std::move(other.task_source_);
  80. return *this;
  81. }
  82. bool JobHandle::IsActive() const {
  83. return task_source_->IsActive();
  84. }
  85. void JobHandle::UpdatePriority(TaskPriority new_priority) {
  86. if (!internal::PooledTaskRunnerDelegate::MatchesCurrentDelegate(
  87. task_source_->delegate())) {
  88. return;
  89. }
  90. task_source_->delegate()->UpdateJobPriority(task_source_, new_priority);
  91. }
  92. void JobHandle::NotifyConcurrencyIncrease() {
  93. if (!internal::PooledTaskRunnerDelegate::MatchesCurrentDelegate(
  94. task_source_->delegate())) {
  95. return;
  96. }
  97. task_source_->NotifyConcurrencyIncrease();
  98. }
  99. void JobHandle::Join() {
  100. DCHECK(internal::PooledTaskRunnerDelegate::MatchesCurrentDelegate(
  101. task_source_->delegate()));
  102. DCHECK_GE(internal::GetTaskPriorityForCurrentThread(),
  103. task_source_->priority_racy())
  104. << "Join may not be called on Job with higher priority than the current "
  105. "thread.";
  106. UpdatePriority(internal::GetTaskPriorityForCurrentThread());
  107. if (task_source_->GetRemainingConcurrency() != 0) {
  108. // Make sure the task source is in the queue if not enough workers are
  109. // contributing. This is necessary for CreateJob(...).Join(). This is a
  110. // noop if the task source was already in the queue.
  111. task_source_->delegate()->EnqueueJobTaskSource(task_source_);
  112. }
  113. bool must_run = task_source_->WillJoin();
  114. while (must_run)
  115. must_run = task_source_->RunJoinTask();
  116. // Remove |task_source_| from the ThreadPool to prevent access to
  117. // |max_concurrency_callback| after Join().
  118. task_source_->delegate()->RemoveJobTaskSource(task_source_);
  119. task_source_ = nullptr;
  120. }
  121. void JobHandle::Cancel() {
  122. DCHECK(internal::PooledTaskRunnerDelegate::MatchesCurrentDelegate(
  123. task_source_->delegate()));
  124. task_source_->Cancel();
  125. bool must_run = task_source_->WillJoin();
  126. DCHECK(!must_run);
  127. // Remove |task_source_| from the ThreadPool to prevent access to
  128. // |max_concurrency_callback| after Join().
  129. task_source_->delegate()->RemoveJobTaskSource(task_source_);
  130. task_source_ = nullptr;
  131. }
  132. void JobHandle::CancelAndDetach() {
  133. task_source_->Cancel();
  134. Detach();
  135. }
  136. void JobHandle::Detach() {
  137. DCHECK(task_source_);
  138. task_source_ = nullptr;
  139. }
  140. JobHandle PostJob(const Location& from_here,
  141. const TaskTraits& traits,
  142. RepeatingCallback<void(JobDelegate*)> worker_task,
  143. MaxConcurrencyCallback max_concurrency_callback) {
  144. auto task_source =
  145. CreateJobTaskSource(from_here, traits, std::move(worker_task),
  146. std::move(max_concurrency_callback));
  147. const bool queued =
  148. static_cast<internal::ThreadPoolImpl*>(ThreadPoolInstance::Get())
  149. ->EnqueueJobTaskSource(task_source);
  150. if (queued)
  151. return internal::JobTaskSource::CreateJobHandle(std::move(task_source));
  152. return JobHandle();
  153. }
  154. JobHandle CreateJob(const Location& from_here,
  155. const TaskTraits& traits,
  156. RepeatingCallback<void(JobDelegate*)> worker_task,
  157. MaxConcurrencyCallback max_concurrency_callback) {
  158. auto task_source =
  159. CreateJobTaskSource(from_here, traits, std::move(worker_task),
  160. std::move(max_concurrency_callback));
  161. return internal::JobTaskSource::CreateJobHandle(std::move(task_source));
  162. }
  163. } // namespace base