Support re-entrant calls to MessageQueueManager::Clear.

webrtc/rtc_base/messagequeue.cc

 /*
  *  Copyright 2004 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 <algorithm>
 
 #include "webrtc/rtc_base/atomicops.h"
 #include "webrtc/rtc_base/checks.h"
 #include "webrtc/rtc_base/logging.h"
 #include "webrtc/rtc_base/messagequeue.h"
 #include "webrtc/rtc_base/stringencode.h"
 #include "webrtc/rtc_base/thread.h"
 #include "webrtc/rtc_base/trace_event.h"
 
 namespace rtc {
 namespace {
 
 const int kMaxMsgLatency = 150;  // 150 ms
 const int kSlowDispatchLoggingThreshold = 50;  // 50 ms
 
-class SCOPED_LOCKABLE DebugNonReentrantCritScope {
+class SCOPED_LOCKABLE MarkProcessingCritScope {
  public:
-  DebugNonReentrantCritScope(const CriticalSection* cs, bool* locked)
+  MarkProcessingCritScope(const CriticalSection* cs, size_t* processing)
       EXCLUSIVE_LOCK_FUNCTION(cs)
-      : cs_(cs), locked_(locked) {
+      : cs_(cs), processing_(processing) {
     cs_->Enter();
-    RTC_DCHECK(!*locked_);
-    *locked_ = true;
+    *processing_ += 1;
   }
 
-  ~DebugNonReentrantCritScope() UNLOCK_FUNCTION() {
-    *locked_ = false;
+  ~MarkProcessingCritScope() UNLOCK_FUNCTION() {
+    *processing_ -= 1;
     cs_->Leave();
   }
 
  private:
   const CriticalSection* const cs_;
-  bool* locked_;
+  size_t* processing_;
 
-  RTC_DISALLOW_COPY_AND_ASSIGN(DebugNonReentrantCritScope);
+  RTC_DISALLOW_COPY_AND_ASSIGN(MarkProcessingCritScope);
 };
 
 class FunctorPostMessageHandler : public MessageHandler {
  public:
   void OnMessage(Message* msg) override {
     RunnableData* data = static_cast<RunnableData*>(msg->pdata);
     data->Run();
     delete data;
   }
 };
@@ skipping 10 matching lines @@
   // spawned.
   if (!instance_)
     instance_ = new MessageQueueManager;
   return instance_;
 }
 
 bool MessageQueueManager::IsInitialized() {
   return instance_ != nullptr;
 }
 
-MessageQueueManager::MessageQueueManager() : locked_(false) {}
+MessageQueueManager::MessageQueueManager() : processing_(0) {}
 
 MessageQueueManager::~MessageQueueManager() {
 }
 
 void MessageQueueManager::Add(MessageQueue *message_queue) {
   return Instance()->AddInternal(message_queue);
 }
 void MessageQueueManager::AddInternal(MessageQueue *message_queue) {
-  DebugNonReentrantCritScope cs(&crit_, &locked_);
+  CritScope cs(&crit_);
+  // Prevent changes while the list of message queues is processed.
+  RTC_DCHECK_EQ(processing_, 0);
   message_queues_.push_back(message_queue);
 }
 
 void MessageQueueManager::Remove(MessageQueue *message_queue) {
   // If there isn't a message queue manager instance, then there isn't a queue
   // to remove.
   if (!instance_) return;
   return Instance()->RemoveInternal(message_queue);
 }
 void MessageQueueManager::RemoveInternal(MessageQueue *message_queue) {
   // If this is the last MessageQueue, destroy the manager as well so that
   // we don't leak this object at program shutdown. As mentioned above, this is
   // not thread-safe, but this should only happen at program termination (when
   // the ThreadManager is destroyed, and threads are no longer active).
   bool destroy = false;
   {
-    DebugNonReentrantCritScope cs(&crit_, &locked_);
+    CritScope cs(&crit_);
+    // Prevent changes while the list of message queues is processed.
+    RTC_DCHECK_EQ(processing_, 0);
     std::vector<MessageQueue *>::iterator iter;
     iter = std::find(message_queues_.begin(), message_queues_.end(),
                      message_queue);
     if (iter != message_queues_.end()) {
       message_queues_.erase(iter);
     }
     destroy = message_queues_.empty();
   }
   if (destroy) {
     instance_ = nullptr;
     delete this;
   }
 }
 
 void MessageQueueManager::Clear(MessageHandler *handler) {
   // If there isn't a message queue manager instance, then there aren't any
   // queues to remove this handler from.
   if (!instance_) return;
   return Instance()->ClearInternal(handler);
 }
 void MessageQueueManager::ClearInternal(MessageHandler *handler) {
-  DebugNonReentrantCritScope cs(&crit_, &locked_);
+  // Deleted objects may cause re-entrant calls to ClearInternal. This is
+  // allowed as the list of message queues does not change while queues are
+  // cleared.
+  MarkProcessingCritScope cs(&crit_, &processing_);
   std::vector<MessageQueue *>::iterator iter;
-  for (iter = message_queues_.begin(); iter != message_queues_.end(); iter++)
-    (*iter)->Clear(handler);
+  for (MessageQueue* queue : message_queues_) {
+    queue->Clear(handler);
+  }
 }
 
 void MessageQueueManager::ProcessAllMessageQueues() {
   if (!instance_) {
     return;
   }
   return Instance()->ProcessAllMessageQueuesInternal();
 }
 
 void MessageQueueManager::ProcessAllMessageQueuesInternal() {
   // This works by posting a delayed message at the current time and waiting
   // for it to be dispatched on all queues, which will ensure that all messages
   // that came before it were also dispatched.
   volatile int queues_not_done = 0;
 
   // This class is used so that whether the posted message is processed, or the
   // message queue is simply cleared, queues_not_done gets decremented.
   class ScopedIncrement : public MessageData {
    public:
     ScopedIncrement(volatile int* value) : value_(value) {
       AtomicOps::Increment(value_);
     }
     ~ScopedIncrement() override { AtomicOps::Decrement(value_); }
 
    private:
     volatile int* value_;
   };
 
   {
-    DebugNonReentrantCritScope cs(&crit_, &locked_);
+    MarkProcessingCritScope cs(&crit_, &processing_);
     for (MessageQueue* queue : message_queues_) {
       if (!queue->IsProcessingMessages()) {
         // If the queue is not processing messages, it can
         // be ignored. If we tried to post a message to it, it would be dropped
         // or ignored.
         continue;
       }
       queue->PostDelayed(RTC_FROM_HERE, 0, nullptr, MQID_DISPOSE,
                          new ScopedIncrement(&queues_not_done));
     }
@@ skipping 373 matching lines @@
 
 void MessageQueue::PostFunctorInternal(const Location& posted_from,
                                        RunnableData* message_data) {
   // Use static to ensure it outlives this scope. Safe since
   // FunctorPostMessageHandler keeps no state.
   static FunctorPostMessageHandler handler;
   Post(posted_from, &handler, 0, message_data);
 }
 
 }  // namespace rtc