Improving the fake clock and using it to fix a flaky STUN timeout test.

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>
 
+#include "webrtc/base/atomicops.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/common.h"
 #include "webrtc/base/logging.h"
 #include "webrtc/base/messagequeue.h"
+#include "webrtc/base/thread.h"
 #include "webrtc/base/trace_event.h"
 
-namespace {
-
-enum { MSG_WAKE_MESSAGE_QUEUE = 1 };
-}
-
 namespace rtc {
 
 const int kMaxMsgLatency = 150;  // 150 ms
 
 //------------------------------------------------------------------
 // MessageQueueManager
 
 MessageQueueManager* MessageQueueManager::instance_ = NULL;
 
 MessageQueueManager* MessageQueueManager::Instance() {
   // Note: This is not thread safe, but it is first called before threads are
   // spawned.
   if (!instance_)
     instance_ = new MessageQueueManager;
   return instance_;
 }
 
 bool MessageQueueManager::IsInitialized() {
   return instance_ != NULL;
 }
 
-MessageQueueManager::MessageQueueManager() {
-}
+MessageQueueManager::MessageQueueManager() {}
 
 MessageQueueManager::~MessageQueueManager() {
 }
 
 void MessageQueueManager::Add(MessageQueue *message_queue) {
   return Instance()->AddInternal(message_queue);
 }
 void MessageQueueManager::AddInternal(MessageQueue *message_queue) {
   // MessageQueueManager methods should be non-reentrant, so we
   // ASSERT that is the case.  If any of these ASSERT, please
@@ skipping 45 matching lines @@
 void MessageQueueManager::ClearInternal(MessageHandler *handler) {
 #if CS_DEBUG_CHECKS  // CurrentThreadIsOwner returns true by default.
   ASSERT(!crit_.CurrentThreadIsOwner());  // See note above.
 #endif
   CritScope cs(&crit_);
   std::vector<MessageQueue *>::iterator iter;
   for (iter = message_queues_.begin(); iter != message_queues_.end(); iter++)
     (*iter)->Clear(handler);
 }
 
-void MessageQueueManager::WakeAllMessageQueues() {
+void MessageQueueManager::ProcessAllMessageQueues() {
   if (!instance_) {
     return;
   }
-  return Instance()->WakeAllMessageQueuesInternal();
+  return Instance()->ProcessAllMessageQueuesInternal();
 }
 
-void MessageQueueManager::WakeAllMessageQueuesInternal() {
+void MessageQueueManager::ProcessAllMessageQueuesInternal() {
 #if CS_DEBUG_CHECKS  // CurrentThreadIsOwner returns true by default.
   ASSERT(!crit_.CurrentThreadIsOwner());  // See note above.
 #endif
-  CritScope cs(&crit_);
-  for (MessageQueue* queue : message_queues_) {
-    // Posting an arbitrary message will force the message queue to wake up.
-    queue->Post(this, MSG_WAKE_MESSAGE_QUEUE);
+  // Post a delayed message at the current time and wait 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;
+  auto functor = [&queues_not_done] { AtomicOps::Decrement(&queues_not_done); };
+  FunctorMessageHandler<void, decltype(functor)> handler(functor);
+  {
+    CritScope cs(&crit_);
+    queues_not_done = static_cast<int>(message_queues_.size());
+    for (MessageQueue* queue : message_queues_) {
+      queue->PostDelayed(0, &handler);
+    }
+  }
+  // Note: One of the message queues may have been on this thread, which is why
+  // 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);
   }
 }
 
-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) {
   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
@@ skipping 334 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