metronome_source.cc 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361
  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/metronome_source.h"
  5. #include <memory>
  6. #include <utility>
  7. #include "base/bind.h"
  8. #include "base/containers/flat_set.h"
  9. #include "base/logging.h"
  10. #include "base/memory/ptr_util.h"
  11. #include "base/memory/ref_counted.h"
  12. #include "base/memory/scoped_refptr.h"
  13. #include "base/synchronization/lock.h"
  14. #include "base/synchronization/waitable_event.h"
  15. #include "base/task/sequenced_task_runner.h"
  16. #include "base/task/thread_pool.h"
  17. #include "base/thread_annotations.h"
  18. #include "base/time/time.h"
  19. #include "base/trace_event/typed_macros.h"
  20. #include "third_party/webrtc/api/metronome/metronome.h"
  21. #include "third_party/webrtc/api/task_queue/pending_task_safety_flag.h"
  22. namespace blink {
  23. constexpr base::TimeDelta kMetronomeTick = base::Hertz(64);
  24. namespace {
  25. // Stores a MetronomeSource::ListenerHandle which handles listening to handle
  26. // ticks, and an atomic flag for cancelling the task attached to the listener.
  27. // When a TickListener invokes, it will check that the cancel flag was not set.
  28. // To avoid a race between the cancel flag and `OnTick` being invoked after
  29. // `RemoveListener`, `RemoveListener` needs to be called from
  30. // `listener->OnTickTaskQueue()`. This is the case for the only user -
  31. // webrtc::DecodeSynchronizer.
  32. //
  33. // TODO(http://crbug.com/1253787): Clarify threading requirements of
  34. // webrtc::Metronome, or change interface.
  35. struct HandleWithCancelation {
  36. HandleWithCancelation(
  37. scoped_refptr<MetronomeSource::ListenerHandle> handle,
  38. rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety)
  39. : handle(std::move(handle)), task_safety(std::move(task_safety)) {}
  40. scoped_refptr<MetronomeSource::ListenerHandle> handle;
  41. rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety;
  42. };
  43. void InvokeOnTickOnWebRtcTaskQueue(
  44. webrtc::Metronome::TickListener* listener,
  45. rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety) {
  46. listener->OnTickTaskQueue()->PostTask(webrtc::SafeTask(
  47. std::move(task_safety), [listener] { listener->OnTick(); }));
  48. }
  49. class WebRtcMetronomeAdapter : public webrtc::Metronome {
  50. public:
  51. explicit WebRtcMetronomeAdapter(
  52. scoped_refptr<MetronomeSource> metronome_source)
  53. : metronome_source_(std::move(metronome_source)) {
  54. DCHECK(metronome_source_);
  55. }
  56. ~WebRtcMetronomeAdapter() override { DCHECK(listeners_.empty()); }
  57. // Adds a tick listener to the metronome. Once this method has returned
  58. // OnTick will be invoked on each metronome tick. A listener may
  59. // only be added to the metronome once.
  60. void AddListener(TickListener* listener) override {
  61. DCHECK(listener);
  62. auto task_safety = webrtc::PendingTaskSafetyFlag::Create();
  63. // `listener` can be unretained since the `handle` will be cancelled when
  64. // `listener` is removed.
  65. auto handle = metronome_source_->AddListener(
  66. nullptr, base::BindRepeating(&InvokeOnTickOnWebRtcTaskQueue,
  67. base::Unretained(listener), task_safety));
  68. base::AutoLock auto_lock(lock_);
  69. auto [it, inserted] = listeners_.emplace(
  70. std::piecewise_construct, std::forward_as_tuple(listener),
  71. std::forward_as_tuple(std::move(handle), std::move(task_safety)));
  72. DCHECK(inserted);
  73. }
  74. // Removes the tick listener from the metronome. Once this method has returned
  75. // OnTick will never be called again. This method must not be called from
  76. // within OnTick.
  77. void RemoveListener(TickListener* listener) override {
  78. DCHECK(listener);
  79. scoped_refptr<MetronomeSource::ListenerHandle> handle;
  80. rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety;
  81. {
  82. base::AutoLock auto_lock(lock_);
  83. auto it = listeners_.find(listener);
  84. if (it == listeners_.end()) {
  85. DLOG(WARNING) << __FUNCTION__ << " called with unregistered listener.";
  86. return;
  87. }
  88. handle = std::move(it->second.handle);
  89. task_safety = std::move(it->second.task_safety);
  90. listeners_.erase(listener);
  91. }
  92. task_safety->SetNotAlive();
  93. metronome_source_->RemoveListener(std::move(handle));
  94. }
  95. // Returns the current tick period of the metronome.
  96. webrtc::TimeDelta TickPeriod() const override {
  97. return webrtc::TimeDelta::Micros(MetronomeSource::Tick().InMicroseconds());
  98. }
  99. private:
  100. const scoped_refptr<MetronomeSource> metronome_source_;
  101. base::Lock lock_;
  102. base::flat_map<TickListener*, HandleWithCancelation> listeners_
  103. GUARDED_BY(lock_);
  104. };
  105. } // namespace
  106. MetronomeSource::ListenerHandle::ListenerHandle(
  107. scoped_refptr<MetronomeSource> metronome_source,
  108. scoped_refptr<base::SequencedTaskRunner> task_runner,
  109. base::RepeatingCallback<void()> callback,
  110. base::TimeTicks wakeup_time)
  111. : metronome_source_(std::move(metronome_source)),
  112. task_runner_(std::move(task_runner)),
  113. callback_(std::move(callback)),
  114. wakeup_time_(std::move(wakeup_time)) {}
  115. MetronomeSource::ListenerHandle::~ListenerHandle() = default;
  116. void MetronomeSource::ListenerHandle::SetWakeupTime(
  117. base::TimeTicks wakeup_time) {
  118. metronome_source_->metronome_task_runner_->PostTask(
  119. FROM_HERE,
  120. base::BindOnce(
  121. &MetronomeSource::ListenerHandle::SetWakeUpTimeOnMetronomeTaskRunner,
  122. this, wakeup_time));
  123. }
  124. void MetronomeSource::ListenerHandle::SetWakeUpTimeOnMetronomeTaskRunner(
  125. base::TimeTicks wakeup_time) {
  126. DCHECK(
  127. metronome_source_->metronome_task_runner_->RunsTasksInCurrentSequence());
  128. wakeup_time_ = wakeup_time;
  129. metronome_source_->EnsureNextTickIsScheduled(wakeup_time);
  130. }
  131. void MetronomeSource::ListenerHandle::OnMetronomeTickOnMetronomeTaskRunner(
  132. base::TimeTicks now) {
  133. DCHECK(
  134. metronome_source_->metronome_task_runner_->RunsTasksInCurrentSequence());
  135. if (wakeup_time_.is_max()) {
  136. // This listener is sleeping indefinitely.
  137. return;
  138. }
  139. if (now < wakeup_time_) {
  140. // It is not time for this listener to fire yet, but ensure that the next
  141. // tick is scheduled.
  142. metronome_source_->EnsureNextTickIsScheduled(wakeup_time_);
  143. return;
  144. }
  145. if (!wakeup_time_.is_min()) {
  146. // A wakeup time had been specified (set to anything other than "min").
  147. // Reset the wakeup time to "infinity", meaning SetWakeupTime() has to be
  148. // called again in order to wake up again.
  149. wakeup_time_ = base::TimeTicks::Max();
  150. }
  151. if (task_runner_ == nullptr) {
  152. // Run the task directly if |task_runner_| is null.
  153. MaybeRunCallback();
  154. } else {
  155. // Post to run on target |task_runner_|.
  156. task_runner_->PostTask(
  157. FROM_HERE,
  158. base::BindOnce(&MetronomeSource::ListenerHandle::MaybeRunCallback,
  159. this));
  160. }
  161. }
  162. void MetronomeSource::ListenerHandle::MaybeRunCallback() {
  163. DCHECK(task_runner_ == nullptr || task_runner_->RunsTasksInCurrentSequence());
  164. base::AutoLock auto_lock(is_active_lock_);
  165. if (!is_active_)
  166. return;
  167. callback_.Run();
  168. }
  169. void MetronomeSource::ListenerHandle::Inactivate() {
  170. base::AutoLock auto_lock(is_active_lock_);
  171. is_active_ = false;
  172. }
  173. // static
  174. base::TimeTicks MetronomeSource::Phase() {
  175. return base::TimeTicks();
  176. }
  177. // static
  178. base::TimeDelta MetronomeSource::Tick() {
  179. return kMetronomeTick;
  180. }
  181. // static
  182. base::TimeTicks MetronomeSource::TimeSnappedToNextTick(base::TimeTicks time) {
  183. return time.SnappedToNextTick(MetronomeSource::Phase(),
  184. MetronomeSource::Tick());
  185. }
  186. MetronomeSource::MetronomeSource()
  187. : metronome_task_runner_(
  188. // HIGHEST priority is used to reduce risk of jitter.
  189. base::ThreadPool::CreateSequencedTaskRunner(
  190. {base::TaskPriority::HIGHEST})) {
  191. base::TimeTicks now = base::TimeTicks::Now();
  192. prev_tick_ = MetronomeSource::TimeSnappedToNextTick(now);
  193. if (prev_tick_ > now)
  194. prev_tick_ -= MetronomeSource::Tick();
  195. }
  196. MetronomeSource::~MetronomeSource() {
  197. DCHECK(listeners_.empty());
  198. DCHECK(!next_tick_handle_.IsValid());
  199. }
  200. scoped_refptr<MetronomeSource::ListenerHandle> MetronomeSource::AddListener(
  201. scoped_refptr<base::SequencedTaskRunner> task_runner,
  202. base::RepeatingCallback<void()> callback,
  203. base::TimeTicks wakeup_time) {
  204. // Ref-counting keeps |this| alive until all listeners have been removed.
  205. scoped_refptr<ListenerHandle> listener_handle =
  206. base::MakeRefCounted<ListenerHandle>(this, std::move(task_runner),
  207. std::move(callback), wakeup_time);
  208. metronome_task_runner_->PostTask(
  209. FROM_HERE,
  210. base::BindOnce(&MetronomeSource::AddListenerOnMetronomeTaskRunner,
  211. scoped_refptr<MetronomeSource>(this), listener_handle));
  212. return listener_handle;
  213. }
  214. void MetronomeSource::RemoveListener(
  215. scoped_refptr<ListenerHandle> listener_handle) {
  216. listener_handle->Inactivate();
  217. metronome_task_runner_->PostTask(
  218. FROM_HERE,
  219. base::BindOnce(&MetronomeSource::RemoveListenerOnMetronomeTaskRunner,
  220. scoped_refptr<MetronomeSource>(this), listener_handle));
  221. }
  222. void MetronomeSource::AddListenerOnMetronomeTaskRunner(
  223. scoped_refptr<ListenerHandle> listener_handle) {
  224. DCHECK(metronome_task_runner_->RunsTasksInCurrentSequence());
  225. listeners_.insert(listener_handle);
  226. EnsureNextTickIsScheduled(listener_handle->wakeup_time_);
  227. }
  228. void MetronomeSource::RemoveListenerOnMetronomeTaskRunner(
  229. scoped_refptr<ListenerHandle> listener_handle) {
  230. DCHECK(metronome_task_runner_->RunsTasksInCurrentSequence());
  231. listeners_.erase(listener_handle);
  232. // To avoid additional complexity we do not reschedule the next tick, but we
  233. // do cancel the next tick if there are no more listeners.
  234. if (listeners_.empty()) {
  235. next_tick_ = base::TimeTicks::Min();
  236. next_tick_handle_.CancelTask();
  237. }
  238. }
  239. void MetronomeSource::EnsureNextTickIsScheduled(base::TimeTicks wakeup_time) {
  240. DCHECK(metronome_task_runner_->RunsTasksInCurrentSequence());
  241. if (wakeup_time.is_max()) {
  242. return;
  243. }
  244. if (wakeup_time <= prev_tick_) {
  245. // Do not reschedule a tick that already fired, such as when adding a
  246. // listener on a tick.
  247. wakeup_time = prev_tick_ + MetronomeSource::Tick();
  248. }
  249. base::TimeTicks wakeup_tick =
  250. MetronomeSource::TimeSnappedToNextTick(wakeup_time);
  251. if (!next_tick_.is_min() && wakeup_tick >= next_tick_) {
  252. // We already have the next tick scheduled.
  253. return;
  254. }
  255. // If we already have a tick scheduled but too far in the future, cancel it.
  256. next_tick_handle_.CancelTask();
  257. next_tick_ = wakeup_tick;
  258. next_tick_handle_ = metronome_task_runner_->PostCancelableDelayedTaskAt(
  259. base::subtle::PostDelayedTaskPassKey(), FROM_HERE,
  260. base::BindOnce(&MetronomeSource::OnMetronomeTick,
  261. // Unretained is safe because tasks are cancelled prior to
  262. // destruction.
  263. base::Unretained(this), next_tick_),
  264. next_tick_, base::subtle::DelayPolicy::kPrecise);
  265. }
  266. void MetronomeSource::OnMetronomeTick(base::TimeTicks now_tick) {
  267. TRACE_EVENT_INSTANT0("webrtc", "MetronomeSource::OnMetronomeTick",
  268. TRACE_EVENT_SCOPE_PROCESS);
  269. DCHECK(metronome_task_runner_->RunsTasksInCurrentSequence());
  270. // We no longer have a tick scheduled.
  271. prev_tick_ = now_tick;
  272. next_tick_ = base::TimeTicks::Min();
  273. bool schedule_next_tick = false;
  274. base::TimeTicks now = base::TimeTicks::Now();
  275. // On some platforms (Android), base::TimeTicks::Now() may in some cases lag
  276. // behind by ~1 ms due to clock caching. To ensure listeners with the same
  277. // wake up time as |now_tick| runs, ensure |now| is at least |now_tick|.
  278. if (now < now_tick) {
  279. now = now_tick;
  280. }
  281. for (auto& listener : listeners_) {
  282. listener->OnMetronomeTickOnMetronomeTaskRunner(now);
  283. schedule_next_tick |= listener->wakeup_time_.is_min();
  284. }
  285. if (schedule_next_tick) {
  286. // The next tick is `now_tick + metronome_tick_`, but if late wakeup happens
  287. // due to load we could in extreme cases miss ticks. To avoid posting
  288. // immediate "catch-up" tasks, make it possible to skip metronome ticks.
  289. constexpr double kTickThreshold = 0.5;
  290. EnsureNextTickIsScheduled(base::TimeTicks::Now() +
  291. MetronomeSource::Tick() * kTickThreshold);
  292. }
  293. }
  294. std::unique_ptr<webrtc::Metronome> MetronomeSource::CreateWebRtcMetronome() {
  295. return std::make_unique<WebRtcMetronomeAdapter>(base::WrapRefCounted(this));
  296. }
  297. base::TimeDelta MetronomeSource::EnsureNextTickAndGetDelayForTesting() {
  298. base::TimeTicks now = base::TimeTicks::Now();
  299. base::TimeTicks next_tick = MetronomeSource::TimeSnappedToNextTick(now);
  300. // Ensure next tick is scheduled, even if there are no listeners. This makes
  301. // it so that when mock time is advanced to |next_tick|, |prev_tick_| will be
  302. // updated. This avoids the initial tick firing "now" in testing environments.
  303. metronome_task_runner_->PostTask(
  304. FROM_HERE, base::BindOnce(&MetronomeSource::EnsureNextTickIsScheduled,
  305. this, next_tick));
  306. return next_tick - now;
  307. }
  308. bool MetronomeSource::HasListenersForTesting() {
  309. base::WaitableEvent event;
  310. bool has_listeners = false;
  311. metronome_task_runner_->PostTask(
  312. FROM_HERE, base::BindOnce(
  313. [](MetronomeSource* thiz, bool* has_listeners,
  314. base::WaitableEvent* event) {
  315. *has_listeners = !thiz->listeners_.empty();
  316. event->Signal();
  317. },
  318. base::Unretained(this), base::Unretained(&has_listeners),
  319. base::Unretained(&event)));
  320. event.Wait();
  321. return has_listeners;
  322. }
  323. } // namespace blink