task_queue_factory.cc 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. // Copyright 2021 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 "third_party/webrtc_overrides/task_queue_factory.h"
  5. #include <map>
  6. #include <memory>
  7. #include "base/check.h"
  8. #include "base/logging.h"
  9. #include "base/memory/ref_counted.h"
  10. #include "base/task/thread_pool.h"
  11. #include "base/thread_annotations.h"
  12. #include "base/time/time.h"
  13. #include "third_party/abseil-cpp/absl/functional/any_invocable.h"
  14. #include "third_party/webrtc/api/task_queue/task_queue_base.h"
  15. #include "third_party/webrtc/api/task_queue/task_queue_factory.h"
  16. #include "third_party/webrtc/api/units/time_delta.h"
  17. #include "third_party/webrtc_overrides/coalesced_tasks.h"
  18. #include "third_party/webrtc_overrides/metronome_source.h"
  19. namespace blink {
  20. class WebRtcTaskQueue : public webrtc::TaskQueueBase {
  21. public:
  22. explicit WebRtcTaskQueue(base::TaskTraits traits);
  23. // webrtc::TaskQueueBase implementation.
  24. void Delete() override;
  25. void PostTask(absl::AnyInvocable<void() &&> task) override;
  26. void PostDelayedTask(absl::AnyInvocable<void() &&> task,
  27. webrtc::TimeDelta delay) override;
  28. void PostDelayedHighPrecisionTask(absl::AnyInvocable<void() &&> task,
  29. webrtc::TimeDelta delay) override;
  30. private:
  31. // Runs a single PostTask-task.
  32. static void MaybeRunTask(WebRtcTaskQueue* task_queue,
  33. scoped_refptr<base::RefCountedData<bool>> is_active,
  34. absl::AnyInvocable<void() &&> task);
  35. void RunTask(absl::AnyInvocable<void() &&> task);
  36. // Runs all ready PostDelayedTask-tasks that have been scheduled to run at
  37. // |scheduled_time_now|.
  38. static void MaybeRunCoalescedTasks(
  39. WebRtcTaskQueue* task_queue,
  40. scoped_refptr<base::RefCountedData<bool>> is_active,
  41. base::TimeTicks scheduled_time_now);
  42. const scoped_refptr<base::SequencedTaskRunner> task_runner_;
  43. // Value of |is_active_| is checked and set on |task_runner_|.
  44. const scoped_refptr<base::RefCountedData<bool>> is_active_;
  45. // Low precision tasks are coalesced onto metronome ticks and stored in
  46. // |coalesced_tasks_| until they are ready to run.
  47. CoalescedTasks coalesced_tasks_;
  48. };
  49. WebRtcTaskQueue::WebRtcTaskQueue(base::TaskTraits traits)
  50. : task_runner_(
  51. base::ThreadPool::CreateSequencedTaskRunner(std::move(traits))),
  52. is_active_(new base::RefCountedData<bool>(true)) {}
  53. void Deactivate(scoped_refptr<base::RefCountedData<bool>> is_active,
  54. CoalescedTasks* coalesced_tasks,
  55. base::WaitableEvent* event) {
  56. is_active->data = false;
  57. coalesced_tasks->Clear();
  58. event->Signal();
  59. }
  60. void WebRtcTaskQueue::Delete() {
  61. // Ensure there are no in-flight PostTask-tasks when deleting.
  62. base::WaitableEvent event;
  63. task_runner_->PostTask(FROM_HERE, base::BindOnce(&Deactivate, is_active_,
  64. &coalesced_tasks_, &event));
  65. event.Wait();
  66. delete this;
  67. }
  68. void WebRtcTaskQueue::PostTask(absl::AnyInvocable<void() &&> task) {
  69. // Delete() ensures there are no in-flight tasks at destruction, so passing an
  70. // unretained pointer to |this| is safe.
  71. task_runner_->PostTask(
  72. FROM_HERE, base::BindOnce(&WebRtcTaskQueue::RunTask,
  73. base::Unretained(this), std::move(task)));
  74. }
  75. // static
  76. void WebRtcTaskQueue::MaybeRunTask(
  77. WebRtcTaskQueue* task_queue,
  78. scoped_refptr<base::RefCountedData<bool>> is_active,
  79. absl::AnyInvocable<void() &&> task) {
  80. if (!is_active->data)
  81. return;
  82. task_queue->RunTask(std::move(task));
  83. }
  84. void WebRtcTaskQueue::RunTask(absl::AnyInvocable<void() &&> task) {
  85. CurrentTaskQueueSetter set_current(this);
  86. std::move(task)();
  87. }
  88. // static
  89. void WebRtcTaskQueue::MaybeRunCoalescedTasks(
  90. WebRtcTaskQueue* task_queue,
  91. scoped_refptr<base::RefCountedData<bool>> is_active,
  92. base::TimeTicks scheduled_time_now) {
  93. if (!is_active->data)
  94. return;
  95. CurrentTaskQueueSetter set_current(task_queue);
  96. task_queue->coalesced_tasks_.RunScheduledTasks(scheduled_time_now);
  97. }
  98. void WebRtcTaskQueue::PostDelayedTask(absl::AnyInvocable<void() &&> task,
  99. webrtc::TimeDelta delay) {
  100. base::TimeTicks target_time =
  101. base::TimeTicks::Now() + base::Microseconds(delay.us());
  102. base::TimeTicks snapped_target_time =
  103. MetronomeSource::TimeSnappedToNextTick(target_time);
  104. // Queue to run the delayed task at |snapped_target_time|. If the snapped time
  105. // has not been scheduled before, schedule it with PostDelayedTaskAt().
  106. if (coalesced_tasks_.QueueDelayedTask(target_time, std::move(task),
  107. snapped_target_time)) {
  108. // The posted task might outlive |this|, but access to |this| is guarded by
  109. // the ref-counted |is_active_| flag.
  110. task_runner_->PostDelayedTaskAt(
  111. base::subtle::PostDelayedTaskPassKey(), FROM_HERE,
  112. base::BindOnce(&WebRtcTaskQueue::MaybeRunCoalescedTasks,
  113. base::Unretained(this), is_active_, snapped_target_time),
  114. snapped_target_time, base::subtle::DelayPolicy::kPrecise);
  115. }
  116. }
  117. void WebRtcTaskQueue::PostDelayedHighPrecisionTask(
  118. absl::AnyInvocable<void() &&> task,
  119. webrtc::TimeDelta delay) {
  120. base::TimeTicks target_time =
  121. base::TimeTicks::Now() + base::Microseconds(delay.us());
  122. // The posted task might outlive |this|, but access to |this| is guarded by
  123. // the ref-counted |is_active_| flag.
  124. task_runner_->PostDelayedTaskAt(
  125. base::subtle::PostDelayedTaskPassKey(), FROM_HERE,
  126. base::BindOnce(&WebRtcTaskQueue::MaybeRunTask, base::Unretained(this),
  127. is_active_, std::move(task)),
  128. target_time, base::subtle::DelayPolicy::kPrecise);
  129. }
  130. namespace {
  131. base::TaskTraits TaskQueuePriority2Traits(
  132. webrtc::TaskQueueFactory::Priority priority) {
  133. // The content/renderer/media/webrtc/rtc_video_encoder.* code
  134. // employs a PostTask/Wait pattern that uses TQ in a way that makes it
  135. // blocking and synchronous, which is why we allow WithBaseSyncPrimitives()
  136. // for OS_ANDROID.
  137. switch (priority) {
  138. case webrtc::TaskQueueFactory::Priority::HIGH:
  139. #if defined(OS_ANDROID)
  140. return {base::WithBaseSyncPrimitives(), base::TaskPriority::HIGHEST};
  141. #else
  142. return {base::TaskPriority::HIGHEST};
  143. #endif
  144. case webrtc::TaskQueueFactory::Priority::LOW:
  145. return {base::MayBlock(), base::TaskPriority::BEST_EFFORT};
  146. case webrtc::TaskQueueFactory::Priority::NORMAL:
  147. default:
  148. #if defined(OS_ANDROID)
  149. return {base::WithBaseSyncPrimitives()};
  150. #else
  151. return {};
  152. #endif
  153. }
  154. }
  155. class WebrtcTaskQueueFactory final : public webrtc::TaskQueueFactory {
  156. public:
  157. std::unique_ptr<webrtc::TaskQueueBase, webrtc::TaskQueueDeleter>
  158. CreateTaskQueue(absl::string_view name, Priority priority) const override {
  159. return std::unique_ptr<webrtc::TaskQueueBase, webrtc::TaskQueueDeleter>(
  160. new WebRtcTaskQueue(TaskQueuePriority2Traits(priority)));
  161. }
  162. };
  163. } // namespace
  164. } // namespace blink
  165. std::unique_ptr<webrtc::TaskQueueFactory> CreateWebRtcTaskQueueFactory() {
  166. return std::unique_ptr<webrtc::TaskQueueFactory>(
  167. new blink::WebrtcTaskQueueFactory());
  168. }
  169. std::unique_ptr<webrtc::TaskQueueBase, webrtc::TaskQueueDeleter>
  170. CreateWebRtcTaskQueue(webrtc::TaskQueueFactory::Priority priority) {
  171. return std::unique_ptr<webrtc::TaskQueueBase, webrtc::TaskQueueDeleter>(
  172. new blink::WebRtcTaskQueue(blink::TaskQueuePriority2Traits(priority)));
  173. }