Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(1187)

Side by Side Diff: webrtc/base/messagequeue.cc

Issue 2038213002: Revert of Improving the fake clock and using it to fix a flaky STUN timeout test. (Closed) Base URL: https://chromium.googlesource.com/external/webrtc.git@master
Patch Set: Created 4 years, 6 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
« no previous file with comments | « webrtc/base/messagequeue.h ('k') | webrtc/base/signalthread_unittest.cc » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 /* 1 /*
2 * Copyright 2004 The WebRTC Project Authors. All rights reserved. 2 * Copyright 2004 The WebRTC Project Authors. All rights reserved.
3 * 3 *
4 * Use of this source code is governed by a BSD-style license 4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source 5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found 6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may 7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree. 8 * be found in the AUTHORS file in the root of the source tree.
9 */ 9 */
10 #include <algorithm> 10 #include <algorithm>
11 11
12 #include "webrtc/base/atomicops.h"
13 #include "webrtc/base/checks.h" 12 #include "webrtc/base/checks.h"
14 #include "webrtc/base/common.h" 13 #include "webrtc/base/common.h"
15 #include "webrtc/base/logging.h" 14 #include "webrtc/base/logging.h"
16 #include "webrtc/base/messagequeue.h" 15 #include "webrtc/base/messagequeue.h"
17 #include "webrtc/base/thread.h"
18 #include "webrtc/base/trace_event.h" 16 #include "webrtc/base/trace_event.h"
19 17
18 namespace {
19
20 enum { MSG_WAKE_MESSAGE_QUEUE = 1 };
21 }
22
20 namespace rtc { 23 namespace rtc {
21 24
22 const int kMaxMsgLatency = 150; // 150 ms 25 const int kMaxMsgLatency = 150; // 150 ms
23 26
24 //------------------------------------------------------------------ 27 //------------------------------------------------------------------
25 // MessageQueueManager 28 // MessageQueueManager
26 29
27 MessageQueueManager* MessageQueueManager::instance_ = NULL; 30 MessageQueueManager* MessageQueueManager::instance_ = NULL;
28 31
29 MessageQueueManager* MessageQueueManager::Instance() { 32 MessageQueueManager* MessageQueueManager::Instance() {
30 // Note: This is not thread safe, but it is first called before threads are 33 // Note: This is not thread safe, but it is first called before threads are
31 // spawned. 34 // spawned.
32 if (!instance_) 35 if (!instance_)
33 instance_ = new MessageQueueManager; 36 instance_ = new MessageQueueManager;
34 return instance_; 37 return instance_;
35 } 38 }
36 39
37 bool MessageQueueManager::IsInitialized() { 40 bool MessageQueueManager::IsInitialized() {
38 return instance_ != NULL; 41 return instance_ != NULL;
39 } 42 }
40 43
41 MessageQueueManager::MessageQueueManager() {} 44 MessageQueueManager::MessageQueueManager() {
45 }
42 46
43 MessageQueueManager::~MessageQueueManager() { 47 MessageQueueManager::~MessageQueueManager() {
44 } 48 }
45 49
46 void MessageQueueManager::Add(MessageQueue *message_queue) { 50 void MessageQueueManager::Add(MessageQueue *message_queue) {
47 return Instance()->AddInternal(message_queue); 51 return Instance()->AddInternal(message_queue);
48 } 52 }
49 void MessageQueueManager::AddInternal(MessageQueue *message_queue) { 53 void MessageQueueManager::AddInternal(MessageQueue *message_queue) {
50 // MessageQueueManager methods should be non-reentrant, so we 54 // MessageQueueManager methods should be non-reentrant, so we
51 // ASSERT that is the case. If any of these ASSERT, please 55 // ASSERT that is the case. If any of these ASSERT, please
(...skipping 45 matching lines...) Expand 10 before | Expand all | Expand 10 after
97 void MessageQueueManager::ClearInternal(MessageHandler *handler) { 101 void MessageQueueManager::ClearInternal(MessageHandler *handler) {
98 #if CS_DEBUG_CHECKS // CurrentThreadIsOwner returns true by default. 102 #if CS_DEBUG_CHECKS // CurrentThreadIsOwner returns true by default.
99 ASSERT(!crit_.CurrentThreadIsOwner()); // See note above. 103 ASSERT(!crit_.CurrentThreadIsOwner()); // See note above.
100 #endif 104 #endif
101 CritScope cs(&crit_); 105 CritScope cs(&crit_);
102 std::vector<MessageQueue *>::iterator iter; 106 std::vector<MessageQueue *>::iterator iter;
103 for (iter = message_queues_.begin(); iter != message_queues_.end(); iter++) 107 for (iter = message_queues_.begin(); iter != message_queues_.end(); iter++)
104 (*iter)->Clear(handler); 108 (*iter)->Clear(handler);
105 } 109 }
106 110
107 void MessageQueueManager::ProcessAllMessageQueues() { 111 void MessageQueueManager::WakeAllMessageQueues() {
108 if (!instance_) { 112 if (!instance_) {
109 return; 113 return;
110 } 114 }
111 return Instance()->ProcessAllMessageQueuesInternal(); 115 return Instance()->WakeAllMessageQueuesInternal();
112 } 116 }
113 117
114 void MessageQueueManager::ProcessAllMessageQueuesInternal() { 118 void MessageQueueManager::WakeAllMessageQueuesInternal() {
115 #if CS_DEBUG_CHECKS // CurrentThreadIsOwner returns true by default. 119 #if CS_DEBUG_CHECKS // CurrentThreadIsOwner returns true by default.
116 ASSERT(!crit_.CurrentThreadIsOwner()); // See note above. 120 ASSERT(!crit_.CurrentThreadIsOwner()); // See note above.
117 #endif 121 #endif
118 // Post a delayed message at the current time and wait for it to be dispatched 122 CritScope cs(&crit_);
119 // on all queues, which will ensure that all messages that came before it were 123 for (MessageQueue* queue : message_queues_) {
120 // also dispatched. 124 // Posting an arbitrary message will force the message queue to wake up.
121 volatile int queues_not_done; 125 queue->Post(this, MSG_WAKE_MESSAGE_QUEUE);
122 auto functor = [&queues_not_done] { AtomicOps::Decrement(&queues_not_done); };
123 FunctorMessageHandler<void, decltype(functor)> handler(functor);
124 {
125 CritScope cs(&crit_);
126 queues_not_done = static_cast<int>(message_queues_.size());
127 for (MessageQueue* queue : message_queues_) {
128 queue->PostDelayed(0, &handler);
129 }
130 }
131 // Note: One of the message queues may have been on this thread, which is why
132 // we can't synchronously wait for queues_not_done to go to 0; we need to
133 // process messages as well.
134 while (AtomicOps::AcquireLoad(&queues_not_done) > 0) {
135 rtc::Thread::Current()->ProcessMessages(0);
136 } 126 }
137 } 127 }
138 128
129 void MessageQueueManager::OnMessage(Message* pmsg) {
130 RTC_DCHECK(pmsg->message_id == MSG_WAKE_MESSAGE_QUEUE);
131 }
132
139 //------------------------------------------------------------------ 133 //------------------------------------------------------------------
140 // MessageQueue 134 // MessageQueue
141 MessageQueue::MessageQueue(SocketServer* ss, bool init_queue) 135 MessageQueue::MessageQueue(SocketServer* ss, bool init_queue)
142 : fStop_(false), fPeekKeep_(false), 136 : fStop_(false), fPeekKeep_(false),
143 dmsgq_next_num_(0), fInitialized_(false), fDestroyed_(false), ss_(ss) { 137 dmsgq_next_num_(0), fInitialized_(false), fDestroyed_(false), ss_(ss) {
144 RTC_DCHECK(ss); 138 RTC_DCHECK(ss);
145 // Currently, MessageQueue holds a socket server, and is the base class for 139 // Currently, MessageQueue holds a socket server, and is the base class for
146 // Thread. It seems like it makes more sense for Thread to hold the socket 140 // Thread. It seems like it makes more sense for Thread to hold the socket
147 // server, and provide it to the MessageQueue, since the Thread controls 141 // server, and provide it to the MessageQueue, since the Thread controls
148 // the I/O model, and MQ is agnostic to those details. Anyway, this causes 142 // the I/O model, and MQ is agnostic to those details. Anyway, this causes
(...skipping 334 matching lines...) Expand 10 before | Expand all | Expand 10 after
483 dmsgq_.container().erase(new_end, dmsgq_.container().end()); 477 dmsgq_.container().erase(new_end, dmsgq_.container().end());
484 dmsgq_.reheap(); 478 dmsgq_.reheap();
485 } 479 }
486 480
487 void MessageQueue::Dispatch(Message *pmsg) { 481 void MessageQueue::Dispatch(Message *pmsg) {
488 TRACE_EVENT0("webrtc", "MessageQueue::Dispatch"); 482 TRACE_EVENT0("webrtc", "MessageQueue::Dispatch");
489 pmsg->phandler->OnMessage(pmsg); 483 pmsg->phandler->OnMessage(pmsg);
490 } 484 }
491 485
492 } // namespace rtc 486 } // namespace rtc
OLDNEW
« no previous file with comments | « webrtc/base/messagequeue.h ('k') | webrtc/base/signalthread_unittest.cc » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698