Metronome: disable & refactor for single-threaded operation.

The Chromium implementation unfortunately has a rare deadlock.
Rather than patching that up, we're changing the metronome
implementation to be able to use a single-threaded environment
instead.

The metronome functionality is disabled in VideoReceiveStream2
construction inside call.cc.

The new design does not have listener registration or
deresigstration and instead accepts and invokes callbacks, on
the same sequence that requested the callback. This allows
the clients to use features such as WeakPtrFactories or
ScopedThreadSafety for cancellation.

The CL will be followed up with cleanup CLs that removes
registration APIs once downstream consumers have adapted.

Bug: chromium:1381982
Change-Id: I43732d1971e2276c39b431a04365cd2fc3c55c25
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/282280
Reviewed-by: Per Kjellander <perkj@webrtc.org>
Reviewed-by: Erik Språng <sprang@webrtc.org>
Reviewed-by: Evan Shrubsole <eshr@webrtc.org>
Commit-Queue: Markus Handell <handellm@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#38582}

api/metronome/BUILD.gn

@@ -10,7 +10,10 @@ import("../../webrtc.gni")
 
 rtc_source_set("metronome") {
   visibility = [ "*" ]
-  sources = [ "metronome.h" ]
+  sources = [
+    "metronome.cc",
+    "metronome.h",
+  ]
   deps = [
     "../../rtc_base/system:rtc_export",
     "../task_queue",

api/metronome/metronome.cc

@@ -0,0 +1,19 @@
+/*
+ *  Copyright (c) 2022 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#include "api/metronome/metronome.h"
+
+namespace webrtc {
+
+// TODO(crbug.com/1381982): Remove outdated methods.
+void Metronome::AddListener(TickListener* listener) {}
+void Metronome::RemoveListener(TickListener* listener) {}
+
+}  // namespace webrtc

api/metronome/metronome.h

@@ -17,44 +17,40 @@
 
 namespace webrtc {
 
-// The Metronome posts OnTick() on task queues provided by its listeners' task
-// queue periodically. The metronome can be used as an alternative to using
-// PostDelayedTask on a thread or task queue for coalescing work and reducing
-// the number of idle-wakeups.
-//
-// Listeners can be added and removed from any sequence, but it is illegal to
-// remove a listener from an OnTick invocation.
+// The Metronome posts OnTick() calls requested with RequestCallOnNextTick.
+// The API is designed to be fully used from a single task queue. Scheduled
+// callbacks are executed on the same sequence as they were requested on. There
+// are no features implemented for cancellation. When that's needed, use e.g.
+// ScopedTaskSafety from the client.
 //
 // The metronome concept is still under experimentation, and may not be availble
 // in all platforms or applications. See https://crbug.com/1253787 for more
 // details.
 //
-// Metronome implementations must be thread-safe.
+// Metronome implementations must be thread-compatible.
 class RTC_EXPORT Metronome {
  public:
+  // TODO(crbug.com/1381982): remove stale classes and methods once downstream
+  // dependencies adapts.
   class RTC_EXPORT TickListener {
    public:
     virtual ~TickListener() = default;
-
-    // OnTick is run on the task queue provided by OnTickTaskQueue each time the
-    // metronome ticks.
     virtual void OnTick() = 0;
-
-    // The task queue that OnTick will run on. Must not be null.
     virtual TaskQueueBase* OnTickTaskQueue() = 0;
   };
 
   virtual ~Metronome() = default;
 
-  // Adds a tick listener to the metronome. Once this method has returned
-  // OnTick will be invoked on each metronome tick. A listener may
-  // only be added to the metronome once.
-  virtual void AddListener(TickListener* listener) = 0;
+  // TODO(crbug.com/1381982): remove stale classes and methods once downstream
+  // dependencies adapts.
+  virtual void AddListener(TickListener* listener);
+  virtual void RemoveListener(TickListener* listener);
 
-  // Removes the tick listener from the metronome. Once this method has returned
-  // OnTick will never be called again. This method must not be called from
-  // within OnTick.
-  virtual void RemoveListener(TickListener* listener) = 0;
+  // Requests a call to `callback` on the next tick. Scheduled callbacks are
+  // executed on the same sequence as they were requested on. There are no
+  // features for cancellation. When that's needed, use e.g. ScopedTaskSafety
+  // from the client.
+  virtual void RequestCallOnNextTick(absl::AnyInvocable<void() &&> callback) {}
 
   // Returns the current tick period of the metronome.
   virtual TimeDelta TickPeriod() const = 0;

api/metronome/test/BUILD.gn

@@ -23,6 +23,7 @@ rtc_library("fake_metronome") {
     "../../../rtc_base:rtc_task_queue",
     "../../../rtc_base/synchronization:mutex",
     "../../../rtc_base/task_utils:repeating_task",
+    "../../../test:test_support",
     "../../task_queue",
     "../../units:time_delta",
   ]

api/metronome/test/fake_metronome.cc

@@ -10,8 +10,12 @@
 
 #include "api/metronome/test/fake_metronome.h"
 
+#include <utility>
+#include <vector>
+
 #include "api/priority.h"
 #include "api/sequence_checker.h"
+#include "api/task_queue/task_queue_base.h"
 #include "api/task_queue/task_queue_factory.h"
 #include "api/units/time_delta.h"
 #include "rtc_base/event.h"
@@ -22,12 +26,9 @@ namespace webrtc::test {
 ForcedTickMetronome::ForcedTickMetronome(TimeDelta tick_period)
     : tick_period_(tick_period) {}
 
-void ForcedTickMetronome::AddListener(TickListener* listener) {
-  listeners_.insert(listener);
-}
-
-void ForcedTickMetronome::RemoveListener(TickListener* listener) {
-  listeners_.erase(listener);
+void ForcedTickMetronome::RequestCallOnNextTick(
+    absl::AnyInvocable<void() &&> callback) {
+  callbacks_.push_back(std::move(callback));
 }
 
 TimeDelta ForcedTickMetronome::TickPeriod() const {
@@ -35,55 +36,35 @@ TimeDelta ForcedTickMetronome::TickPeriod() const {
 }
 
 size_t ForcedTickMetronome::NumListeners() {
-  return listeners_.size();
+  return callbacks_.size();
 }
 
 void ForcedTickMetronome::Tick() {
-  for (auto* listener : listeners_) {
-    listener->OnTickTaskQueue()->PostTask([listener] { listener->OnTick(); });
+  std::vector<absl::AnyInvocable<void() &&>> callbacks;
+  callbacks_.swap(callbacks);
+  for (auto& callback : callbacks)
+    std::move(callback)();
+}
+
+FakeMetronome::FakeMetronome(TimeDelta tick_period)
+    : tick_period_(tick_period) {}
+
+void FakeMetronome::RequestCallOnNextTick(
+    absl::AnyInvocable<void() &&> callback) {
+  TaskQueueBase* current = TaskQueueBase::Current();
+  callbacks_.push_back(std::move(callback));
+  if (callbacks_.size() == 1) {
+    current->PostDelayedTask(
+        [this] {
+          std::vector<absl::AnyInvocable<void() &&>> callbacks;
+          callbacks_.swap(callbacks);
+          for (auto& callback : callbacks)
+            std::move(callback)();
+        },
+        tick_period_);
   }
 }
 
-FakeMetronome::FakeMetronome(TaskQueueFactory* factory, TimeDelta tick_period)
-    : tick_period_(tick_period),
-      queue_(factory->CreateTaskQueue("MetronomeQueue",
-                                      TaskQueueFactory::Priority::HIGH)) {}
-
-FakeMetronome::~FakeMetronome() {
-  RTC_DCHECK(listeners_.empty());
-}
-
-void FakeMetronome::AddListener(TickListener* listener) {
-  MutexLock lock(&mutex_);
-  listeners_.insert(listener);
-  if (!started_) {
-    tick_task_ = RepeatingTaskHandle::Start(queue_.Get(), [this] {
-      MutexLock lock(&mutex_);
-      // Stop if empty.
-      if (listeners_.empty())
-        return TimeDelta::PlusInfinity();
-      for (auto* listener : listeners_) {
-        listener->OnTickTaskQueue()->PostTask(
-            [listener] { listener->OnTick(); });
-      }
-      return tick_period_;
-    });
-    started_ = true;
-  }
-}
-
-void FakeMetronome::RemoveListener(TickListener* listener) {
-  MutexLock lock(&mutex_);
-  listeners_.erase(listener);
-}
-
-void FakeMetronome::Stop() {
-  MutexLock lock(&mutex_);
-  RTC_DCHECK(listeners_.empty());
-  if (started_)
-    queue_.PostTask([this] { tick_task_.Stop(); });
-}
-
 TimeDelta FakeMetronome::TickPeriod() const {
   return tick_period_;
 }

api/metronome/test/fake_metronome.h

@@ -13,6 +13,7 @@
 
 #include <memory>
 #include <set>
+#include <vector>
 
 #include "api/metronome/metronome.h"
 #include "api/task_queue/task_queue_base.h"
@@ -36,13 +37,12 @@ class ForcedTickMetronome : public Metronome {
   size_t NumListeners();
 
   // Metronome implementation.
-  void AddListener(TickListener* listener) override;
-  void RemoveListener(TickListener* listener) override;
+  void RequestCallOnNextTick(absl::AnyInvocable<void() &&> callback) override;
   TimeDelta TickPeriod() const override;
 
  private:
   const TimeDelta tick_period_;
-  std::set<TickListener*> listeners_;
+  std::vector<absl::AnyInvocable<void() &&>> callbacks_;
 };
 
 // FakeMetronome is a metronome that ticks based on a repeating task at the
@@ -53,23 +53,15 @@ class ForcedTickMetronome : public Metronome {
 // on the proper task queue.
 class FakeMetronome : public Metronome {
  public:
-  FakeMetronome(TaskQueueFactory* factory, TimeDelta tick_period);
-  ~FakeMetronome() override;
+  explicit FakeMetronome(TimeDelta tick_period);
 
   // Metronome implementation.
-  void AddListener(TickListener* listener) override;
-  void RemoveListener(TickListener* listener) override;
+  void RequestCallOnNextTick(absl::AnyInvocable<void() &&> callback) override;
   TimeDelta TickPeriod() const override;
 
-  void Stop();
-
  private:
   const TimeDelta tick_period_;
-  RepeatingTaskHandle tick_task_;
-  bool started_ RTC_GUARDED_BY(mutex_) = false;
-  std::set<TickListener*> listeners_ RTC_GUARDED_BY(mutex_);
-  Mutex mutex_;
-  rtc::TaskQueue queue_;
+  std::vector<absl::AnyInvocable<void() &&>> callbacks_;
 };
 
 }  // namespace webrtc::test