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 123 matching lines @@
   // we can't synchronously wait for queues_not_done to go to 0; we need to
   // process messages as well.
   while (AtomicOps::AcquireLoad(&queues_not_done) > 0) {
     rtc::Thread::Current()->ProcessMessages(0);
   }
 }
 
 //------------------------------------------------------------------
 // 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),

[Taylor Brandstetter, 2016/07/07 22:01:47]

Since it's an int now, should probably initialize with 0 instead of false.

[andresp, 2016/07/08 10:05:37]

Done.

+      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;
+  AtomicOps::ReleaseStore(&stop_, 1);
   WakeUpSocketServer();
 }
 
 bool MessageQueue::IsQuitting() {
-  return fStop_;
+  return AtomicOps::AcquireLoad(&stop_) != 0;
 }
 
 void MessageQueue::Restart() {
-  fStop_ = false;
+  AtomicOps::ReleaseStore(&stop_, 0);
 }
 
 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 20 matching lines @@
 }
 
 void MessageQueue::ReceiveSends() {
 }
 
 void MessageQueue::Post(const Location& posted_from,
                         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.posted_from = posted_from;
@@ skipping 35 matching lines @@
   return DoDelayPost(posted_from, TimeUntil(tstamp), tstamp, phandler, id,
                      pdata);
 }
 
 void MessageQueue::DoDelayPost(const Location& posted_from,
                                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 85 matching lines @@
   pmsg->phandler->OnMessage(pmsg);
   int64_t end_time = TimeMillis();
   int64_t diff = TimeDiff(end_time, start_time);
   if (diff >= kSlowDispatchLoggingThreshold) {
     LOG(LS_INFO) << "Message took " << diff << "ms to dispatch. Posted from: "
                  << pmsg->posted_from.ToString();
   }
 }
 
 }  // namespace rtc