Add RtcpTransceiver::Stop to allow non-blocking destruction

As downside it disallows to destroy RtcpTransceiver on the TaskQueue
without prio call to the Stop function

BUG: webrtc:8239
Change-Id: I236b9aff7a0746044dd764c619174cc5ac03d27f
Reviewed-on: https://webrtc-review.googlesource.com/98120
Commit-Queue: Danil Chapovalov <danilchap@webrtc.org>
Reviewed-by: Per Kjellander <perkj@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#24587}

modules/rtp_rtcp/BUILD.gn

@@ -255,7 +255,6 @@ rtc_source_set("rtcp_transceiver") {
     "../../rtc_base:rtc_base_approved",
     "../../rtc_base:rtc_cancelable_task",
     "../../rtc_base:rtc_task_queue",
-    "../../rtc_base:weak_ptr",
     "../../system_wrappers",
     "//third_party/abseil-cpp/absl/memory",
     "//third_party/abseil-cpp/absl/types:optional",

modules/rtp_rtcp/source/rtcp_transceiver.cc

@@ -22,41 +22,47 @@ namespace webrtc {
 
 RtcpTransceiver::RtcpTransceiver(const RtcpTransceiverConfig& config)
     : task_queue_(config.task_queue),
-      rtcp_transceiver_(absl::make_unique<RtcpTransceiverImpl>(config)),
+      rtcp_transceiver_(absl::make_unique<RtcpTransceiverImpl>(config)) {
-      ptr_factory_(rtcp_transceiver_.get()),
-      // Creating first weak ptr can be done on any thread, but is not
-      // thread-safe, thus do it at construction. Creating second (e.g. making a
-      // copy) is thread-safe.
-      ptr_(ptr_factory_.GetWeakPtr()) {
   RTC_DCHECK(task_queue_);
 }
 
 RtcpTransceiver::~RtcpTransceiver() {
-  if (task_queue_->IsCurrent())
+  if (!rtcp_transceiver_)
     return;
+  RTC_CHECK(!task_queue_->IsCurrent());
 
   rtc::Event done(false, false);
   // TODO(danilchap): Merge cleanup into main closure when task queue does not
   // silently drop tasks.
   task_queue_->PostTask(rtc::NewClosure(
       [this] {
-        // Destructor steps that has to run on the task_queue_.
+        // Destructor steps that better run on the task_queue_.
-        ptr_factory_.InvalidateWeakPtrs();
         rtcp_transceiver_.reset();
       },
       /*cleanup=*/[&done] { done.Set(); }));
-  // Wait until destruction is complete to be sure weak pointers invalidated and
+  // Wait until destruction is complete to guarantee callbacks are not used
-  // rtcp_transceiver destroyed on the queue while |this| still valid.
+  // after destructor returns.
   done.Wait(rtc::Event::kForever);
   RTC_CHECK(!rtcp_transceiver_) << "Task queue is too busy to handle rtcp";
 }
 
+void RtcpTransceiver::Stop(std::unique_ptr<rtc::QueuedTask> on_destroyed) {
+  RTC_DCHECK(rtcp_transceiver_);
+  struct Destructor {
+    void operator()() { rtcp_transceiver = nullptr; }
+    std::unique_ptr<RtcpTransceiverImpl> rtcp_transceiver;
+  };
+  task_queue_->PostTaskAndReply(Destructor{std::move(rtcp_transceiver_)},
+                                std::move(on_destroyed));
+  RTC_DCHECK(!rtcp_transceiver_);
+}
+
 void RtcpTransceiver::AddMediaReceiverRtcpObserver(
     uint32_t remote_ssrc,
     MediaReceiverRtcpObserver* observer) {
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr = ptr_;
+  RTC_CHECK(rtcp_transceiver_);
+  RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
   task_queue_->PostTask([ptr, remote_ssrc, observer] {
-    if (ptr)
     ptr->AddMediaReceiverRtcpObserver(remote_ssrc, observer);
   });
 }
@@ -65,59 +71,60 @@ void RtcpTransceiver::RemoveMediaReceiverRtcpObserver(
     uint32_t remote_ssrc,
     MediaReceiverRtcpObserver* observer,
     std::unique_ptr<rtc::QueuedTask> on_removed) {
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr = ptr_;
+  RTC_CHECK(rtcp_transceiver_);
+  RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
   auto remove = [ptr, remote_ssrc, observer] {
-    if (ptr)
     ptr->RemoveMediaReceiverRtcpObserver(remote_ssrc, observer);
   };
   task_queue_->PostTaskAndReply(std::move(remove), std::move(on_removed));
 }
 
 void RtcpTransceiver::SetReadyToSend(bool ready) {
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr = ptr_;
+  RTC_CHECK(rtcp_transceiver_);
+  RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
   task_queue_->PostTask([ptr, ready] {
-    if (ptr)
       ptr->SetReadyToSend(ready);
   });
 }
 
 void RtcpTransceiver::ReceivePacket(rtc::CopyOnWriteBuffer packet) {
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr = ptr_;
+  RTC_CHECK(rtcp_transceiver_);
+  RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
   int64_t now_us = rtc::TimeMicros();
   task_queue_->PostTask([ptr, packet, now_us] {
-    if (ptr)
       ptr->ReceivePacket(packet, now_us);
   });
 }
 
 void RtcpTransceiver::SendCompoundPacket() {
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr = ptr_;
+  RTC_CHECK(rtcp_transceiver_);
+  RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
   task_queue_->PostTask([ptr] {
-    if (ptr)
       ptr->SendCompoundPacket();
   });
 }
 
 void RtcpTransceiver::SetRemb(int64_t bitrate_bps,
                               std::vector<uint32_t> ssrcs) {
+  RTC_CHECK(rtcp_transceiver_);
   // TODO(danilchap): Replace with lambda with move capture when available.
   struct SetRembClosure {
     void operator()() {
-      if (ptr)
         ptr->SetRemb(bitrate_bps, std::move(ssrcs));
     }
 
-    rtc::WeakPtr<RtcpTransceiverImpl> ptr;
+    RtcpTransceiverImpl* ptr;
     int64_t bitrate_bps;
     std::vector<uint32_t> ssrcs;
   };
-  task_queue_->PostTask(SetRembClosure{ptr_, bitrate_bps, std::move(ssrcs)});
+  task_queue_->PostTask(
+      SetRembClosure{rtcp_transceiver_.get(), bitrate_bps, std::move(ssrcs)});
 }
 
 void RtcpTransceiver::UnsetRemb() {
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr = ptr_;
+  RTC_CHECK(rtcp_transceiver_);
+  RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
   task_queue_->PostTask([ptr] {
-    if (ptr)
       ptr->UnsetRemb();
   });
 }
@@ -128,54 +135,53 @@ uint32_t RtcpTransceiver::SSRC() const {
 
 bool RtcpTransceiver::SendFeedbackPacket(
     const rtcp::TransportFeedback& packet) {
+  RTC_CHECK(rtcp_transceiver_);
   struct Closure {
     void operator()() {
-      if (ptr)
         ptr->SendRawPacket(raw_packet);
     }
-    rtc::WeakPtr<RtcpTransceiverImpl> ptr;
+    RtcpTransceiverImpl* ptr;
     rtc::Buffer raw_packet;
   };
-  task_queue_->PostTask(Closure{ptr_, packet.Build()});
+  task_queue_->PostTask(Closure{rtcp_transceiver_.get(), packet.Build()});
   return true;
 }
 
 void RtcpTransceiver::SendNack(uint32_t ssrc,
                                std::vector<uint16_t> sequence_numbers) {
+  RTC_CHECK(rtcp_transceiver_);
   // TODO(danilchap): Replace with lambda with move capture when available.
   struct Closure {
     void operator()() {
-      if (ptr)
         ptr->SendNack(ssrc, std::move(sequence_numbers));
     }
 
-    rtc::WeakPtr<RtcpTransceiverImpl> ptr;
+    RtcpTransceiverImpl* ptr;
     uint32_t ssrc;
     std::vector<uint16_t> sequence_numbers;
   };
-  task_queue_->PostTask(Closure{ptr_, ssrc, std::move(sequence_numbers)});
+  task_queue_->PostTask(
+      Closure{rtcp_transceiver_.get(), ssrc, std::move(sequence_numbers)});
 }
 
 void RtcpTransceiver::SendPictureLossIndication(uint32_t ssrc) {
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr = ptr_;
+  RTC_CHECK(rtcp_transceiver_);
-  task_queue_->PostTask([ptr, ssrc] {
+  RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
-    if (ptr)
+  task_queue_->PostTask([ptr, ssrc] { ptr->SendPictureLossIndication(ssrc); });
-      ptr->SendPictureLossIndication(ssrc);
-  });
 }
 
 void RtcpTransceiver::SendFullIntraRequest(std::vector<uint32_t> ssrcs) {
+  RTC_CHECK(rtcp_transceiver_);
   // TODO(danilchap): Replace with lambda with move capture when available.
   struct Closure {
     void operator()() {
-      if (ptr)
         ptr->SendFullIntraRequest(ssrcs);
     }
 
-    rtc::WeakPtr<RtcpTransceiverImpl> ptr;
+    RtcpTransceiverImpl* ptr;
     std::vector<uint32_t> ssrcs;
   };
-  task_queue_->PostTask(Closure{ptr_, std::move(ssrcs)});
+  task_queue_->PostTask(Closure{rtcp_transceiver_.get(), std::move(ssrcs)});
 }
 
 }  // namespace webrtc

modules/rtp_rtcp/source/rtcp_transceiver.h

@@ -20,7 +20,6 @@
 #include "rtc_base/constructormagic.h"
 #include "rtc_base/copyonwritebuffer.h"
 #include "rtc_base/task_queue.h"
-#include "rtc_base/weak_ptr.h"
 
 namespace webrtc {
 //
@@ -30,8 +29,18 @@ namespace webrtc {
 class RtcpTransceiver : public RtcpFeedbackSenderInterface {
  public:
   explicit RtcpTransceiver(const RtcpTransceiverConfig& config);
+  // Blocks unless Stop was called.
+  // TODO(danilchap): Change destructor to never block by breaking assumption
+  // callbacks are not used after destruction.
   ~RtcpTransceiver() override;
 
+  // Start asynchronious destruction of the RtcpTransceiver.
+  // It is safe to call destructor right after Stop exits.
+  // No other methods can be called.
+  // Note that observers provided in constructor or registered with AddObserver
+  // still might be used by the transceiver until |on_destroyed| runs.
+  void Stop(std::unique_ptr<rtc::QueuedTask> on_destroyed);
+
   // Registers observer to be notified about incoming rtcp packets.
   // Calls to observer will be done on the |config.task_queue|.
   void AddMediaReceiverRtcpObserver(uint32_t remote_ssrc,
@@ -80,12 +89,6 @@ class RtcpTransceiver : public RtcpFeedbackSenderInterface {
  private:
   rtc::TaskQueue* const task_queue_;
   std::unique_ptr<RtcpTransceiverImpl> rtcp_transceiver_;
-  rtc::WeakPtrFactory<RtcpTransceiverImpl> ptr_factory_;
-  // TaskQueue, and thus tasks posted to it, may outlive this.
-  // Thus when Posting task class always pass copy of the weak_ptr to access
-  // the RtcpTransceiver and never guarantee it still will be alive when task
-  // runs.
-  rtc::WeakPtr<RtcpTransceiverImpl> ptr_;
 
   RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(RtcpTransceiver);
 };

modules/rtp_rtcp/source/rtcp_transceiver_unittest.cc

@@ -83,18 +83,39 @@ TEST(RtcpTransceiverTest, SendsRtcpOnTaskQueueWhenCreatedOnTaskQueue) {
   WaitPostedTasks(&queue);
 }
 
-TEST(RtcpTransceiverTest, CanBeDestoryedOnTaskQueue) {
+TEST(RtcpTransceiverTest, CanBeDestroyedOnTaskQueueAfterStop) {
   rtc::TaskQueue queue("rtcp");
   NiceMock<MockTransport> outgoing_transport;
   RtcpTransceiverConfig config;
   config.outgoing_transport = &outgoing_transport;
   config.task_queue = &queue;
   auto rtcp_transceiver = absl::make_unique<RtcpTransceiver>(config);
+  rtcp_transceiver->Stop(rtc::NewClosure([] {}));
 
   queue.PostTask([&] { rtcp_transceiver.reset(); });
   WaitPostedTasks(&queue);
 }
 
+TEST(RtcpTransceiverTest, CanBeDestroyedWithoutBlockingAfterStop) {
+  rtc::TaskQueue queue("rtcp");
+  NiceMock<MockTransport> outgoing_transport;
+  RtcpTransceiverConfig config;
+  config.outgoing_transport = &outgoing_transport;
+  config.task_queue = &queue;
+  auto rtcp_transceiver = absl::make_unique<RtcpTransceiver>(config);
+  rtcp_transceiver->SendCompoundPacket();
+
+  rtc::Event heavy_task(false, false);
+  queue.PostTask(
+      rtc::NewClosure([&] { EXPECT_TRUE(heavy_task.Wait(kTimeoutMs)); }));
+  rtc::Event done(false, false);
+  rtcp_transceiver->Stop(rtc::NewClosure([&done] { done.Set(); }));
+  rtcp_transceiver = nullptr;
+
+  heavy_task.Set();
+  EXPECT_TRUE(done.Wait(kTimeoutMs));
+}
+
 // Use rtp timestamp to distinguish different incoming sender reports.
 rtc::CopyOnWriteBuffer CreateSenderReport(uint32_t ssrc, uint32_t rtp_time) {
   webrtc::rtcp::SenderReport sr;
@@ -189,26 +210,23 @@ TEST(RtcpTransceiverTest, CanCallSendCompoundPacketFromAnyThread) {
   WaitPostedTasks(&queue);
 }
 
-TEST(RtcpTransceiverTest, DoesntSendPacketsAfterDestruction) {
+TEST(RtcpTransceiverTest, DoesntSendPacketsAfterStopCallback) {
-  MockTransport outgoing_transport;
+  NiceMock<MockTransport> outgoing_transport;
   rtc::TaskQueue queue("rtcp");
   RtcpTransceiverConfig config;
   config.outgoing_transport = &outgoing_transport;
   config.task_queue = &queue;
-  config.schedule_periodic_compound_packets = false;
+  config.schedule_periodic_compound_packets = true;
-
-  EXPECT_CALL(outgoing_transport, SendRtcp(_, _)).Times(0);
 
   auto rtcp_transceiver = absl::make_unique<RtcpTransceiver>(config);
-  rtc::Event pause(false, false);
+  rtc::Event done(false, false);
-  queue.PostTask([&] {
-    pause.Wait(rtc::Event::kForever);
-    rtcp_transceiver.reset();
-  });
   rtcp_transceiver->SendCompoundPacket();
-  pause.Set();
+  rtcp_transceiver->Stop(rtc::NewClosure([&] {
-  WaitPostedTasks(&queue);
+    EXPECT_CALL(outgoing_transport, SendRtcp).Times(0);
-  EXPECT_FALSE(rtcp_transceiver);
+    done.Set();
+  }));
+  rtcp_transceiver = nullptr;
+  EXPECT_TRUE(done.Wait(kTimeoutMs));
 }
 
 TEST(RtcpTransceiverTest, SendsTransportFeedbackOnTaskQueue) {