thread_wrapper.h 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194
  1. // Copyright (c) 2012 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 COMPONENTS_WEBRTC_THREAD_WRAPPER_H_
  5. #define COMPONENTS_WEBRTC_THREAD_WRAPPER_H_
  6. #include <stdint.h>
  7. #include <list>
  8. #include <map>
  9. #include <memory>
  10. #include "base/callback_forward.h"
  11. #include "base/compiler_specific.h"
  12. #include "base/feature_list.h"
  13. #include "base/memory/weak_ptr.h"
  14. #include "base/synchronization/lock.h"
  15. #include "base/synchronization/waitable_event.h"
  16. #include "base/task/current_thread.h"
  17. #include "base/task/single_thread_task_runner.h"
  18. #include "base/time/time.h"
  19. #include "third_party/abseil-cpp/absl/types/optional.h"
  20. #include "third_party/webrtc/rtc_base/thread.h"
  21. #include "third_party/webrtc_overrides/coalesced_tasks.h"
  22. namespace webrtc {
  23. // ThreadWrapper implements rtc::Thread interface on top of
  24. // Chromium's SingleThreadTaskRunner interface. Currently only the bare minimum
  25. // that is used by P2P part of libjingle is implemented. There are two ways to
  26. // create this object:
  27. //
  28. // - Call EnsureForCurrentMessageLoop(). This approach works only on threads
  29. // that have MessageLoop In this case ThreadWrapper deletes itself
  30. // automatically when MessageLoop is destroyed.
  31. // - Using ThreadWrapper() constructor. In this case the creating code
  32. // must pass a valid task runner for the current thread and also delete the
  33. // wrapper later.
  34. class ThreadWrapper : public base::CurrentThread::DestructionObserver,
  35. public rtc::Thread {
  36. public:
  37. // A repeating callback whose TimeDelta argument indicates a duration sample.
  38. // What the duration represents is contextual.
  39. using SampledDurationCallback =
  40. base::RepeatingCallback<void(base::TimeDelta)>;
  41. // Create ThreadWrapper for the current thread if it hasn't been created
  42. // yet. The thread wrapper is destroyed automatically when the current
  43. // MessageLoop is destroyed.
  44. static void EnsureForCurrentMessageLoop();
  45. // Creates ThreadWrapper for |task_runner| that runs tasks on the
  46. // current thread.
  47. static std::unique_ptr<ThreadWrapper> WrapTaskRunner(
  48. scoped_refptr<base::SingleThreadTaskRunner> task_runner);
  49. // Returns thread wrapper for the current thread or nullptr if it doesn't
  50. // exist.
  51. static ThreadWrapper* current();
  52. // Sets task latency & duration sample callbacks intended to gather UMA
  53. // statistics. Samples are acquired periodically every several seconds by
  54. // ThreadWrapper. In this context,
  55. // * task latency is defined as the duration between the moment a task is
  56. // scheduled from ThreadWrapper's task runner, and the moment
  57. // it begins running.
  58. // * task duration is defined as the duration between the moment the
  59. // ThreadWrapper begins running a task and the moment it ends
  60. // executing it. It only measures durations of tasks posted to rtc::Thread.
  61. // The passed callbacks are called in the ThreadWrapper's task runner
  62. // context.
  63. void SetLatencyAndTaskDurationCallbacks(
  64. SampledDurationCallback task_latency_callback,
  65. SampledDurationCallback task_duration_callback);
  66. ThreadWrapper(const ThreadWrapper&) = delete;
  67. ThreadWrapper& operator=(const ThreadWrapper&) = delete;
  68. ~ThreadWrapper() override;
  69. // Sets whether the thread can be used to send messages
  70. // synchronously to another thread using Send() method. Set to false
  71. // by default to avoid potential jankiness when Send() used on
  72. // renderer thread. It should be set explicitly for threads that
  73. // need to call Send() for other threads.
  74. void set_send_allowed(bool allowed) { send_allowed_ = allowed; }
  75. rtc::SocketServer* SocketServer();
  76. // CurrentThread::DestructionObserver implementation.
  77. void WillDestroyCurrentMessageLoop() override;
  78. // rtc::MessageQueue overrides.
  79. void Post(const rtc::Location& posted_from,
  80. rtc::MessageHandler* phandler,
  81. uint32_t id,
  82. rtc::MessageData* pdata,
  83. bool time_sensitive) override;
  84. void PostDelayed(const rtc::Location& posted_from,
  85. int delay_ms,
  86. rtc::MessageHandler* handler,
  87. uint32_t id,
  88. rtc::MessageData* data) override;
  89. void Clear(rtc::MessageHandler* handler,
  90. uint32_t id,
  91. rtc::MessageList* removed) override;
  92. void Dispatch(rtc::Message* message) override;
  93. void Send(const rtc::Location& posted_from,
  94. rtc::MessageHandler* handler,
  95. uint32_t id,
  96. rtc::MessageData* data) override;
  97. // Quitting is not supported (see below); this method performs
  98. // NOTIMPLEMENTED_LOG_ONCE() and returns false.
  99. // TODO(https://crbug.com/webrtc/10364): When rtc::MessageQueue::Post()
  100. // returns a bool, !IsQuitting() will not be needed to infer success and we
  101. // may implement this as NOTREACHED() like the rest of the methods.
  102. bool IsQuitting() override;
  103. // Following methods are not supported. They are overriden just to
  104. // ensure that they are not called (each of them contain NOTREACHED
  105. // in the body). Some of this methods can be implemented if it
  106. // becomes necessary to use libjingle code that calls them.
  107. void Quit() override;
  108. void Restart() override;
  109. bool Get(rtc::Message* message, int delay_ms, bool process_io) override;
  110. bool Peek(rtc::Message* message, int delay_ms) override;
  111. int GetDelay() override;
  112. // rtc::Thread overrides.
  113. void Stop() override;
  114. void Run() override;
  115. private:
  116. typedef std::map<int, rtc::Message> MessagesQueue;
  117. struct PendingSend;
  118. class PostTaskLatencySampler;
  119. explicit ThreadWrapper(
  120. scoped_refptr<base::SingleThreadTaskRunner> task_runner);
  121. void PostTaskInternal(const rtc::Location& posted_from,
  122. int delay_ms,
  123. rtc::MessageHandler* handler,
  124. uint32_t message_id,
  125. rtc::MessageData* data);
  126. void RunTask(int task_id);
  127. void RunTaskInternal(int task_id);
  128. void ProcessPendingSends();
  129. // TaskQueueBase overrides.
  130. void PostTask(absl::AnyInvocable<void() &&> task) override;
  131. void PostDelayedTask(absl::AnyInvocable<void() &&> task,
  132. webrtc::TimeDelta delay) override;
  133. void PostDelayedHighPrecisionTask(absl::AnyInvocable<void() &&> task,
  134. webrtc::TimeDelta delay) override;
  135. // Executes WebRTC queued tasks from TaskQueueBase overrides on
  136. // |task_runner_|.
  137. void RunTaskQueueTask(absl::AnyInvocable<void() &&> task);
  138. void RunCoalescedTaskQueueTasks(base::TimeTicks scheduled_time);
  139. // Called before a task runs, returns an opaque optional timestamp which
  140. // should be passed into FinalizeRunTask.
  141. absl::optional<base::TimeTicks> PrepareRunTask();
  142. // Called after a task has run. Move the return value of PrepareRunTask as
  143. // |task_start_timestamp|.
  144. void FinalizeRunTask(absl::optional<base::TimeTicks> task_start_timestamp);
  145. // Task runner used to execute messages posted on this thread.
  146. scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
  147. bool send_allowed_;
  148. // |lock_| must be locked when accessing |messages_|.
  149. base::Lock lock_;
  150. int last_task_id_;
  151. MessagesQueue messages_;
  152. std::list<PendingSend*> pending_send_messages_;
  153. base::WaitableEvent pending_send_event_;
  154. std::unique_ptr<PostTaskLatencySampler> latency_sampler_;
  155. SampledDurationCallback task_latency_callback_;
  156. SampledDurationCallback task_duration_callback_;
  157. // Low precision tasks are coalesced onto metronome ticks and stored in
  158. // `coalesced_tasks_` until they are ready to run.
  159. blink::CoalescedTasks coalesced_tasks_;
  160. base::WeakPtr<ThreadWrapper> weak_ptr_;
  161. base::WeakPtrFactory<ThreadWrapper> weak_ptr_factory_{this};
  162. };
  163. } // namespace webrtc
  164. #endif // COMPONENTS_WEBRTC_THREAD_WRAPPER_H_