Protect MessageQueue stop field with a critical section to avoid data races.

webrtc/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>
@@ skipping 115 matching lines @@
   }
 }
 
 void MessageQueueManager::OnMessage(Message* pmsg) {
   RTC_DCHECK(pmsg->message_id == MSG_WAKE_MESSAGE_QUEUE);
 }
 
 //------------------------------------------------------------------
 // MessageQueue
 MessageQueue::MessageQueue(SocketServer* ss, bool init_queue)
-    : fStop_(false), fPeekKeep_(false),
-      dmsgq_next_num_(0), fInitialized_(false), fDestroyed_(false), ss_(ss) {
+    : fPeekKeep_(false),
+      dmsgq_next_num_(0),
+      fInitialized_(false),
+      fDestroyed_(false),
+      stop_(false),
+      ss_(ss) {
   RTC_DCHECK(ss);
   // Currently, MessageQueue holds a socket server, and is the base class for
   // Thread.  It seems like it makes more sense for Thread to hold the socket
   // server, and provide it to the MessageQueue, since the Thread controls
   // the I/O model, and MQ is agnostic to those details.  Anyway, this causes
   // messagequeue_unittest to depend on network libraries... yuck.
   ss_->SetMessageQueue(this);
   if (init_queue) {
     DoInit();
   }
@@ skipping 51 matching lines @@
   ss_ = ss ? ss : own_ss_.get();
   ss_->SetMessageQueue(this);
 }
 
 void MessageQueue::WakeUpSocketServer() {
   SharedScope ss(&ss_lock_);
   ss_->WakeUp();
 }
 
 void MessageQueue::Quit() {
-  fStop_ = true;
+  {
+    CritScope cs(&stop_crit_);
+    stop_ = true;

[pthatcher1, 2016/06/01 15:29:14]

Might as well rename "stop_" to "quitting_".

+  }
   WakeUpSocketServer();
 }
 
 bool MessageQueue::IsQuitting() {
-  return fStop_;
+  CritScope cs(&stop_crit_);

[pthatcher1, 2016/06/01 15:29:14]

Should we use an std::atomic here once they are allowed?  Perhaps with a comment
similar to the one in thread_unittest.cc above AtomicBool.

[tommi (sloooow) - chröme, 2016/06/01 15:53:29]

We do have alternatives to std::atomic that we can use instead.  See
webrtc/base/atomicops.h.

+  return stop_;
 }
 
 void MessageQueue::Restart() {
-  fStop_ = false;
+  CritScope cs(&stop_crit_);
+  stop_ = false;
 }
 
 bool MessageQueue::Peek(Message *pmsg, int cmsWait) {
   if (fPeekKeep_) {
     *pmsg = msgPeek_;
     return true;
   }
   if (!Get(pmsg, cmsWait))
     return false;
   msgPeek_ = *pmsg;
@@ skipping 63 matching lines @@
       // If this was a dispose message, delete it and skip it.
       if (MQID_DISPOSE == pmsg->message_id) {
         ASSERT(NULL == pmsg->phandler);
         delete pmsg->pdata;
         *pmsg = Message();
         continue;
       }
       return true;
     }
 
-    if (fStop_)
+    if (IsQuitting())
       break;
 
     // Which is shorter, the delay wait or the asked wait?
 
     int64_t cmsNext;
     if (cmsWait == kForever) {
       cmsNext = cmsDelayNext;
     } else {
       cmsNext = std::max<int64_t>(0, cmsTotal - cmsElapsed);
       if ((cmsDelayNext != kForever) && (cmsDelayNext < cmsNext))
@@ skipping 19 matching lines @@
   return false;
 }
 
 void MessageQueue::ReceiveSends() {
 }
 
 void MessageQueue::Post(MessageHandler* phandler,
                         uint32_t id,
                         MessageData* pdata,
                         bool time_sensitive) {
-  if (fStop_)
+  if (IsQuitting())
     return;
 
   // Keep thread safe
   // Add the message to the end of the queue
   // Signal for the multiplexer to return
 
   {
     CritScope cs(&crit_);
     Message msg;
     msg.phandler = phandler;
@@ skipping 28 matching lines @@
                           uint32_t id,
                           MessageData* pdata) {
   return DoDelayPost(TimeUntil(tstamp), tstamp, phandler, id, pdata);
 }
 
 void MessageQueue::DoDelayPost(int64_t cmsDelay,
                                int64_t tstamp,
                                MessageHandler* phandler,
                                uint32_t id,
                                MessageData* pdata) {
-  if (fStop_) {
+  if (IsQuitting()) {
     return;
   }
 
   // Keep thread safe
   // Add to the priority queue. Gets sorted soonest first.
   // Signal for the multiplexer to return.
 
   {
     CritScope cs(&crit_);
     Message msg;
@@ skipping 75 matching lines @@
   dmsgq_.container().erase(new_end, dmsgq_.container().end());
   dmsgq_.reheap();
 }
 
 void MessageQueue::Dispatch(Message *pmsg) {
   TRACE_EVENT0("webrtc", "MessageQueue::Dispatch");
   pmsg->phandler->OnMessage(pmsg);
 }
 
 }  // namespace rtc