Prevent data race in MessageQueue.

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.
  */
 
@@ skipping 99 matching lines @@
   CritScope cs(&crit_);
   std::vector<MessageQueue *>::iterator iter;
   for (iter = message_queues_.begin(); iter != message_queues_.end(); iter++)
     (*iter)->Clear(handler);
 }
 
 //------------------------------------------------------------------
 // MessageQueue
 
 MessageQueue::MessageQueue(SocketServer* ss, bool init_queue)
-    : ss_(ss), fStop_(false), fPeekKeep_(false),
-      dmsgq_next_num_(0), fInitialized_(false), fDestroyed_(false) {
+    : fStop_(false), fPeekKeep_(false),
+      dmsgq_next_num_(0), fInitialized_(false), fDestroyed_(false), ss_(ss) {
   if (!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.
     default_ss_.reset(new DefaultSocketServer());
     ss_ = default_ss_.get();
   }
   ss_->SetMessageQueue(this);
@@ skipping 20 matching lines @@
     return;
   }
 
   fDestroyed_ = true;
   // The signal is done from here to ensure
   // that it always gets called when the queue
   // is going away.
   SignalQueueDestroyed();
   MessageQueueManager::Remove(this);
   Clear(NULL);
+
+  SharedScope ss(&ss_lock_);
   if (ss_) {
     ss_->SetMessageQueue(NULL);
   }
 }
 
+SocketServer* MessageQueue::socketserver() {
+  SharedScope ss(&ss_lock_);
+  return ss_;
+}
+
 void MessageQueue::set_socketserver(SocketServer* ss) {
+  // Need to lock exclusively here to prevent simultaneous modifications from
+  // other threads. Can't be a shared lock to prevent races with other reading
+  // threads.
+  // Other places that only read "ss_" can use a shared lock as simultaneous
+  // read access is allowed.
+  ExclusiveScope es(&ss_lock_);
   ss_ = ss ? ss : default_ss_.get();
   ss_->SetMessageQueue(this);
 }
 
-void MessageQueue::Quit() {
-  fStop_ = true;
+void MessageQueue::WakeUpSocketServer() {
+  SharedScope ss(&ss_lock_);
   ss_->WakeUp();
 }
 
+void MessageQueue::Quit() {
+  fStop_ = true;
+  WakeUpSocketServer();
+}
+
 bool MessageQueue::IsQuitting() {
   return fStop_;
 }
 
 void MessageQueue::Restart() {
   fStop_ = false;
 }
 
 bool MessageQueue::Peek(Message *pmsg, int cmsWait) {
   if (fPeekKeep_) {
@@ skipping 83 matching lines @@
 
     int cmsNext;
     if (cmsWait == kForever) {
       cmsNext = cmsDelayNext;
     } else {
       cmsNext = std::max(0, cmsTotal - cmsElapsed);
       if ((cmsDelayNext != kForever) && (cmsDelayNext < cmsNext))
         cmsNext = cmsDelayNext;
     }
 
-    // Wait and multiplex in the meantime
-    if (!ss_->Wait(cmsNext, process_io))
-      return false;
+    {
+      // Wait and multiplex in the meantime
+      SharedScope ss(&ss_lock_);
+      if (!ss_->Wait(cmsNext, process_io))
+        return false;
+    }
 
     // If the specified timeout expired, return
 
     msCurrent = Time();
     cmsElapsed = TimeDiff(msCurrent, msStart);
     if (cmsWait != kForever) {
       if (cmsElapsed >= cmsWait)
         return false;
     }
   }
   return false;
 }
 
 void MessageQueue::ReceiveSends() {
 }
 
 void MessageQueue::Post(MessageHandler* phandler,
                         uint32_t id,
                         MessageData* pdata,
                         bool time_sensitive) {
   if (fStop_)
     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;
-  msg.message_id = id;
-  msg.pdata = pdata;
-  if (time_sensitive) {
-    msg.ts_sensitive = Time() + kMaxMsgLatency;
+  {
+    CritScope cs(&crit_);
+    Message msg;
+    msg.phandler = phandler;
+    msg.message_id = id;
+    msg.pdata = pdata;
+    if (time_sensitive) {
+      msg.ts_sensitive = Time() + kMaxMsgLatency;
+    }
+    msgq_.push_back(msg);
   }
-  msgq_.push_back(msg);
-  ss_->WakeUp();
+  WakeUpSocketServer();
 }
 
 void MessageQueue::PostDelayed(int cmsDelay,
                                MessageHandler* phandler,
                                uint32_t id,
                                MessageData* pdata) {
   return DoDelayPost(cmsDelay, TimeAfter(cmsDelay), phandler, id, pdata);
 }
 
 void MessageQueue::PostAt(uint32_t tstamp,
                           MessageHandler* phandler,
                           uint32_t id,
                           MessageData* pdata) {
   return DoDelayPost(TimeUntil(tstamp), tstamp, phandler, id, pdata);
 }
 
 void MessageQueue::DoDelayPost(int cmsDelay,
                                uint32_t tstamp,
                                MessageHandler* phandler,
                                uint32_t id,
                                MessageData* pdata) {
   if (fStop_)
     return;
 
   // Keep thread safe
   // Add to the priority queue. Gets sorted soonest first.
   // Signal for the multiplexer to return.
 
-  CritScope cs(&crit_);
-  Message msg;
-  msg.phandler = phandler;
-  msg.message_id = id;
-  msg.pdata = pdata;
-  DelayedMessage dmsg(cmsDelay, tstamp, dmsgq_next_num_, msg);
-  dmsgq_.push(dmsg);
-  // If this message queue processes 1 message every millisecond for 50 days,
-  // we will wrap this number.  Even then, only messages with identical times
-  // will be misordered, and then only briefly.  This is probably ok.
-  VERIFY(0 != ++dmsgq_next_num_);
-  ss_->WakeUp();
+  {
+    CritScope cs(&crit_);
+    Message msg;
+    msg.phandler = phandler;
+    msg.message_id = id;
+    msg.pdata = pdata;
+    DelayedMessage dmsg(cmsDelay, tstamp, dmsgq_next_num_, msg);
+    dmsgq_.push(dmsg);
+    // If this message queue processes 1 message every millisecond for 50 days,
+    // we will wrap this number.  Even then, only messages with identical times
+    // will be misordered, and then only briefly.  This is probably ok.
+    VERIFY(0 != ++dmsgq_next_num_);
+  }
+  WakeUpSocketServer();
 }
 
 int MessageQueue::GetDelay() {
   CritScope cs(&crit_);
 
   if (!msgq_.empty())
     return 0;
 
   if (!dmsgq_.empty()) {
     int delay = TimeUntil(dmsgq_.top().msTrigger_);
@@ skipping 53 matching lines @@
   }
   dmsgq_.container().erase(new_end, dmsgq_.container().end());
   dmsgq_.reheap();
 }
 
 void MessageQueue::Dispatch(Message *pmsg) {
   pmsg->phandler->OnMessage(pmsg);
 }
 
 }  // namespace rtc