blob: 8dd84cb89c256d4b3fd99ab6ba1240ef3300954e [file] [log] [blame]
henrike@webrtc.orgf0488722014-05-13 18:00:26 +00001/*
2 * Copyright 2004 The WebRTC Project Authors. All rights reserved.
3 *
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
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
andresp@webrtc.orgff689be2015-02-12 11:54:26 +000010#include <algorithm>
11
deadbeeff5f03e82016-06-06 11:16:06 -070012#include "webrtc/base/atomicops.h"
danilchapbebf54c2016-04-28 01:32:48 -070013#include "webrtc/base/checks.h"
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000014#include "webrtc/base/logging.h"
15#include "webrtc/base/messagequeue.h"
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -070016#include "webrtc/base/stringencode.h"
deadbeeff5f03e82016-06-06 11:16:06 -070017#include "webrtc/base/thread.h"
pbos79e28422016-04-29 08:48:05 -070018#include "webrtc/base/trace_event.h"
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000019
20namespace rtc {
andrespcdf61722016-07-08 02:45:40 -070021namespace {
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000022
Honghai Zhang82d78622016-05-06 11:29:15 -070023const int kMaxMsgLatency = 150; // 150 ms
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -070024const int kSlowDispatchLoggingThreshold = 50; // 50 ms
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000025
andrespcdf61722016-07-08 02:45:40 -070026class SCOPED_LOCKABLE DebugNonReentrantCritScope {
27 public:
28 DebugNonReentrantCritScope(const CriticalSection* cs, bool* locked)
29 EXCLUSIVE_LOCK_FUNCTION(cs)
30 : cs_(cs), locked_(locked) {
31 cs_->Enter();
nisseede5da42017-01-12 05:15:36 -080032 RTC_DCHECK(!*locked_);
andrespcdf61722016-07-08 02:45:40 -070033 *locked_ = true;
34 }
35
36 ~DebugNonReentrantCritScope() UNLOCK_FUNCTION() {
37 *locked_ = false;
38 cs_->Leave();
39 }
40
41 private:
42 const CriticalSection* const cs_;
43 bool* locked_;
44
45 RTC_DISALLOW_COPY_AND_ASSIGN(DebugNonReentrantCritScope);
46};
47} // namespace
48
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000049//------------------------------------------------------------------
50// MessageQueueManager
51
52MessageQueueManager* MessageQueueManager::instance_ = NULL;
53
54MessageQueueManager* MessageQueueManager::Instance() {
55 // Note: This is not thread safe, but it is first called before threads are
56 // spawned.
57 if (!instance_)
58 instance_ = new MessageQueueManager;
59 return instance_;
60}
61
62bool MessageQueueManager::IsInitialized() {
63 return instance_ != NULL;
64}
65
andrespcdf61722016-07-08 02:45:40 -070066MessageQueueManager::MessageQueueManager() : locked_(false) {}
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000067
68MessageQueueManager::~MessageQueueManager() {
69}
70
71void MessageQueueManager::Add(MessageQueue *message_queue) {
72 return Instance()->AddInternal(message_queue);
73}
74void MessageQueueManager::AddInternal(MessageQueue *message_queue) {
andrespcdf61722016-07-08 02:45:40 -070075 DebugNonReentrantCritScope cs(&crit_, &locked_);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000076 message_queues_.push_back(message_queue);
77}
78
79void MessageQueueManager::Remove(MessageQueue *message_queue) {
80 // If there isn't a message queue manager instance, then there isn't a queue
81 // to remove.
82 if (!instance_) return;
83 return Instance()->RemoveInternal(message_queue);
84}
85void MessageQueueManager::RemoveInternal(MessageQueue *message_queue) {
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000086 // If this is the last MessageQueue, destroy the manager as well so that
87 // we don't leak this object at program shutdown. As mentioned above, this is
88 // not thread-safe, but this should only happen at program termination (when
89 // the ThreadManager is destroyed, and threads are no longer active).
90 bool destroy = false;
91 {
andrespcdf61722016-07-08 02:45:40 -070092 DebugNonReentrantCritScope cs(&crit_, &locked_);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +000093 std::vector<MessageQueue *>::iterator iter;
94 iter = std::find(message_queues_.begin(), message_queues_.end(),
95 message_queue);
96 if (iter != message_queues_.end()) {
97 message_queues_.erase(iter);
98 }
99 destroy = message_queues_.empty();
100 }
101 if (destroy) {
102 instance_ = NULL;
103 delete this;
104 }
105}
106
107void MessageQueueManager::Clear(MessageHandler *handler) {
108 // If there isn't a message queue manager instance, then there aren't any
109 // queues to remove this handler from.
110 if (!instance_) return;
111 return Instance()->ClearInternal(handler);
112}
113void MessageQueueManager::ClearInternal(MessageHandler *handler) {
andrespcdf61722016-07-08 02:45:40 -0700114 DebugNonReentrantCritScope cs(&crit_, &locked_);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000115 std::vector<MessageQueue *>::iterator iter;
116 for (iter = message_queues_.begin(); iter != message_queues_.end(); iter++)
117 (*iter)->Clear(handler);
118}
119
deadbeeff5f03e82016-06-06 11:16:06 -0700120void MessageQueueManager::ProcessAllMessageQueues() {
Taylor Brandstetterb3c68102016-05-27 14:15:43 -0700121 if (!instance_) {
122 return;
123 }
deadbeeff5f03e82016-06-06 11:16:06 -0700124 return Instance()->ProcessAllMessageQueuesInternal();
Taylor Brandstetterb3c68102016-05-27 14:15:43 -0700125}
126
deadbeeff5f03e82016-06-06 11:16:06 -0700127void MessageQueueManager::ProcessAllMessageQueuesInternal() {
Taylor Brandstetterfe7d0912016-09-15 17:47:42 -0700128 // This works by posting a delayed message at the current time and waiting
129 // for it to be dispatched on all queues, which will ensure that all messages
130 // that came before it were also dispatched.
131 volatile int queues_not_done = 0;
132
133 // This class is used so that whether the posted message is processed, or the
134 // message queue is simply cleared, queues_not_done gets decremented.
135 class ScopedIncrement : public MessageData {
136 public:
137 ScopedIncrement(volatile int* value) : value_(value) {
138 AtomicOps::Increment(value_);
139 }
140 ~ScopedIncrement() override { AtomicOps::Decrement(value_); }
141
142 private:
143 volatile int* value_;
144 };
145
deadbeeff5f03e82016-06-06 11:16:06 -0700146 {
andrespcdf61722016-07-08 02:45:40 -0700147 DebugNonReentrantCritScope cs(&crit_, &locked_);
deadbeeff5f03e82016-06-06 11:16:06 -0700148 for (MessageQueue* queue : message_queues_) {
Taylor Brandstetterfe7d0912016-09-15 17:47:42 -0700149 if (queue->IsQuitting()) {
150 // If the queue is quitting, it's done processing messages so it can
151 // be ignored. If we tried to post a message to it, it would be dropped.
152 continue;
153 }
154 queue->PostDelayed(RTC_FROM_HERE, 0, nullptr, MQID_DISPOSE,
155 new ScopedIncrement(&queues_not_done));
deadbeeff5f03e82016-06-06 11:16:06 -0700156 }
Taylor Brandstetterb3c68102016-05-27 14:15:43 -0700157 }
deadbeeff5f03e82016-06-06 11:16:06 -0700158 // Note: One of the message queues may have been on this thread, which is why
159 // we can't synchronously wait for queues_not_done to go to 0; we need to
160 // process messages as well.
161 while (AtomicOps::AcquireLoad(&queues_not_done) > 0) {
162 rtc::Thread::Current()->ProcessMessages(0);
163 }
Taylor Brandstetterb3c68102016-05-27 14:15:43 -0700164}
165
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000166//------------------------------------------------------------------
167// MessageQueue
jbauch25d1f282016-02-05 00:25:02 -0800168MessageQueue::MessageQueue(SocketServer* ss, bool init_queue)
André Susano Pinto02a57972016-07-22 13:30:05 +0200169 : fPeekKeep_(false),
170 dmsgq_next_num_(0),
171 fInitialized_(false),
172 fDestroyed_(false),
173 stop_(0),
174 ss_(ss) {
danilchapbebf54c2016-04-28 01:32:48 -0700175 RTC_DCHECK(ss);
176 // Currently, MessageQueue holds a socket server, and is the base class for
177 // Thread. It seems like it makes more sense for Thread to hold the socket
178 // server, and provide it to the MessageQueue, since the Thread controls
179 // the I/O model, and MQ is agnostic to those details. Anyway, this causes
180 // messagequeue_unittest to depend on network libraries... yuck.
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000181 ss_->SetMessageQueue(this);
jbauch25d1f282016-02-05 00:25:02 -0800182 if (init_queue) {
183 DoInit();
184 }
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000185}
186
danilchapbebf54c2016-04-28 01:32:48 -0700187MessageQueue::MessageQueue(std::unique_ptr<SocketServer> ss, bool init_queue)
188 : MessageQueue(ss.get(), init_queue) {
189 own_ss_ = std::move(ss);
190}
191
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000192MessageQueue::~MessageQueue() {
jbauch25d1f282016-02-05 00:25:02 -0800193 DoDestroy();
194}
195
196void MessageQueue::DoInit() {
197 if (fInitialized_) {
198 return;
199 }
200
201 fInitialized_ = true;
202 MessageQueueManager::Add(this);
203}
204
205void MessageQueue::DoDestroy() {
206 if (fDestroyed_) {
207 return;
208 }
209
210 fDestroyed_ = true;
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000211 // The signal is done from here to ensure
212 // that it always gets called when the queue
213 // is going away.
214 SignalQueueDestroyed();
henrike@webrtc.org99b41622014-05-21 20:42:17 +0000215 MessageQueueManager::Remove(this);
216 Clear(NULL);
jbauch9ccedc32016-02-25 01:14:56 -0800217
218 SharedScope ss(&ss_lock_);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000219 if (ss_) {
220 ss_->SetMessageQueue(NULL);
221 }
222}
223
jbauch9ccedc32016-02-25 01:14:56 -0800224SocketServer* MessageQueue::socketserver() {
225 SharedScope ss(&ss_lock_);
226 return ss_;
227}
228
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000229void MessageQueue::set_socketserver(SocketServer* ss) {
jbauch9ccedc32016-02-25 01:14:56 -0800230 // Need to lock exclusively here to prevent simultaneous modifications from
231 // other threads. Can't be a shared lock to prevent races with other reading
232 // threads.
233 // Other places that only read "ss_" can use a shared lock as simultaneous
234 // read access is allowed.
235 ExclusiveScope es(&ss_lock_);
danilchapbebf54c2016-04-28 01:32:48 -0700236 ss_ = ss ? ss : own_ss_.get();
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000237 ss_->SetMessageQueue(this);
238}
239
jbauch9ccedc32016-02-25 01:14:56 -0800240void MessageQueue::WakeUpSocketServer() {
241 SharedScope ss(&ss_lock_);
242 ss_->WakeUp();
243}
244
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000245void MessageQueue::Quit() {
André Susano Pinto02a57972016-07-22 13:30:05 +0200246 AtomicOps::ReleaseStore(&stop_, 1);
jbauch9ccedc32016-02-25 01:14:56 -0800247 WakeUpSocketServer();
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000248}
249
250bool MessageQueue::IsQuitting() {
André Susano Pinto02a57972016-07-22 13:30:05 +0200251 return AtomicOps::AcquireLoad(&stop_) != 0;
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000252}
253
254void MessageQueue::Restart() {
André Susano Pinto02a57972016-07-22 13:30:05 +0200255 AtomicOps::ReleaseStore(&stop_, 0);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000256}
257
258bool MessageQueue::Peek(Message *pmsg, int cmsWait) {
259 if (fPeekKeep_) {
260 *pmsg = msgPeek_;
261 return true;
262 }
263 if (!Get(pmsg, cmsWait))
264 return false;
265 msgPeek_ = *pmsg;
266 fPeekKeep_ = true;
267 return true;
268}
269
270bool MessageQueue::Get(Message *pmsg, int cmsWait, bool process_io) {
271 // Return and clear peek if present
272 // Always return the peek if it exists so there is Peek/Get symmetry
273
274 if (fPeekKeep_) {
275 *pmsg = msgPeek_;
276 fPeekKeep_ = false;
277 return true;
278 }
279
280 // Get w/wait + timer scan / dispatch + socket / event multiplexer dispatch
281
Honghai Zhang82d78622016-05-06 11:29:15 -0700282 int64_t cmsTotal = cmsWait;
283 int64_t cmsElapsed = 0;
284 int64_t msStart = TimeMillis();
285 int64_t msCurrent = msStart;
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000286 while (true) {
287 // Check for sent messages
288 ReceiveSends();
289
290 // Check for posted events
Honghai Zhang82d78622016-05-06 11:29:15 -0700291 int64_t cmsDelayNext = kForever;
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000292 bool first_pass = true;
293 while (true) {
294 // All queue operations need to be locked, but nothing else in this loop
295 // (specifically handling disposed message) can happen inside the crit.
296 // Otherwise, disposed MessageHandlers will cause deadlocks.
297 {
298 CritScope cs(&crit_);
299 // On the first pass, check for delayed messages that have been
300 // triggered and calculate the next trigger time.
301 if (first_pass) {
302 first_pass = false;
303 while (!dmsgq_.empty()) {
Honghai Zhang82d78622016-05-06 11:29:15 -0700304 if (msCurrent < dmsgq_.top().msTrigger_) {
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000305 cmsDelayNext = TimeDiff(dmsgq_.top().msTrigger_, msCurrent);
306 break;
307 }
308 msgq_.push_back(dmsgq_.top().msg_);
309 dmsgq_.pop();
310 }
311 }
312 // Pull a message off the message queue, if available.
313 if (msgq_.empty()) {
314 break;
315 } else {
316 *pmsg = msgq_.front();
317 msgq_.pop_front();
318 }
319 } // crit_ is released here.
320
321 // Log a warning for time-sensitive messages that we're late to deliver.
322 if (pmsg->ts_sensitive) {
Honghai Zhang82d78622016-05-06 11:29:15 -0700323 int64_t delay = TimeDiff(msCurrent, pmsg->ts_sensitive);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000324 if (delay > 0) {
325 LOG_F(LS_WARNING) << "id: " << pmsg->message_id << " delay: "
326 << (delay + kMaxMsgLatency) << "ms";
327 }
328 }
329 // If this was a dispose message, delete it and skip it.
330 if (MQID_DISPOSE == pmsg->message_id) {
nisseede5da42017-01-12 05:15:36 -0800331 RTC_DCHECK(NULL == pmsg->phandler);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000332 delete pmsg->pdata;
333 *pmsg = Message();
334 continue;
335 }
336 return true;
337 }
338
André Susano Pinto02a57972016-07-22 13:30:05 +0200339 if (IsQuitting())
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000340 break;
341
342 // Which is shorter, the delay wait or the asked wait?
343
Honghai Zhang82d78622016-05-06 11:29:15 -0700344 int64_t cmsNext;
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000345 if (cmsWait == kForever) {
346 cmsNext = cmsDelayNext;
347 } else {
Honghai Zhang82d78622016-05-06 11:29:15 -0700348 cmsNext = std::max<int64_t>(0, cmsTotal - cmsElapsed);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000349 if ((cmsDelayNext != kForever) && (cmsDelayNext < cmsNext))
350 cmsNext = cmsDelayNext;
351 }
352
jbauch9ccedc32016-02-25 01:14:56 -0800353 {
354 // Wait and multiplex in the meantime
355 SharedScope ss(&ss_lock_);
Honghai Zhang82d78622016-05-06 11:29:15 -0700356 if (!ss_->Wait(static_cast<int>(cmsNext), process_io))
jbauch9ccedc32016-02-25 01:14:56 -0800357 return false;
358 }
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000359
360 // If the specified timeout expired, return
361
Honghai Zhang82d78622016-05-06 11:29:15 -0700362 msCurrent = TimeMillis();
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000363 cmsElapsed = TimeDiff(msCurrent, msStart);
364 if (cmsWait != kForever) {
365 if (cmsElapsed >= cmsWait)
366 return false;
367 }
368 }
369 return false;
370}
371
372void MessageQueue::ReceiveSends() {
373}
374
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700375void MessageQueue::Post(const Location& posted_from,
376 MessageHandler* phandler,
Peter Boström0c4e06b2015-10-07 12:23:21 +0200377 uint32_t id,
378 MessageData* pdata,
379 bool time_sensitive) {
André Susano Pinto02a57972016-07-22 13:30:05 +0200380 if (IsQuitting())
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000381 return;
382
383 // Keep thread safe
384 // Add the message to the end of the queue
385 // Signal for the multiplexer to return
386
jbauch9ccedc32016-02-25 01:14:56 -0800387 {
388 CritScope cs(&crit_);
389 Message msg;
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700390 msg.posted_from = posted_from;
jbauch9ccedc32016-02-25 01:14:56 -0800391 msg.phandler = phandler;
392 msg.message_id = id;
393 msg.pdata = pdata;
394 if (time_sensitive) {
Honghai Zhang82d78622016-05-06 11:29:15 -0700395 msg.ts_sensitive = TimeMillis() + kMaxMsgLatency;
jbauch9ccedc32016-02-25 01:14:56 -0800396 }
397 msgq_.push_back(msg);
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000398 }
jbauch9ccedc32016-02-25 01:14:56 -0800399 WakeUpSocketServer();
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000400}
401
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700402void MessageQueue::PostDelayed(const Location& posted_from,
403 int cmsDelay,
kwiberg@webrtc.org67186fe2015-03-09 22:21:53 +0000404 MessageHandler* phandler,
Peter Boström0c4e06b2015-10-07 12:23:21 +0200405 uint32_t id,
kwiberg@webrtc.org67186fe2015-03-09 22:21:53 +0000406 MessageData* pdata) {
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700407 return DoDelayPost(posted_from, cmsDelay, TimeAfter(cmsDelay), phandler, id,
408 pdata);
kwiberg@webrtc.org67186fe2015-03-09 22:21:53 +0000409}
410
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700411void MessageQueue::PostAt(const Location& posted_from,
412 uint32_t tstamp,
kwiberg@webrtc.org67186fe2015-03-09 22:21:53 +0000413 MessageHandler* phandler,
Peter Boström0c4e06b2015-10-07 12:23:21 +0200414 uint32_t id,
kwiberg@webrtc.org67186fe2015-03-09 22:21:53 +0000415 MessageData* pdata) {
Honghai Zhang82d78622016-05-06 11:29:15 -0700416 // This should work even if it is used (unexpectedly).
Taylor Brandstetter2b3bf6b2016-05-19 14:57:31 -0700417 int64_t delay = static_cast<uint32_t>(TimeMillis()) - tstamp;
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700418 return DoDelayPost(posted_from, delay, tstamp, phandler, id, pdata);
Honghai Zhang82d78622016-05-06 11:29:15 -0700419}
420
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700421void MessageQueue::PostAt(const Location& posted_from,
422 int64_t tstamp,
Honghai Zhang82d78622016-05-06 11:29:15 -0700423 MessageHandler* phandler,
424 uint32_t id,
425 MessageData* pdata) {
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700426 return DoDelayPost(posted_from, TimeUntil(tstamp), tstamp, phandler, id,
427 pdata);
kwiberg@webrtc.org67186fe2015-03-09 22:21:53 +0000428}
429
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700430void MessageQueue::DoDelayPost(const Location& posted_from,
431 int64_t cmsDelay,
Honghai Zhang82d78622016-05-06 11:29:15 -0700432 int64_t tstamp,
Peter Boström0c4e06b2015-10-07 12:23:21 +0200433 MessageHandler* phandler,
434 uint32_t id,
435 MessageData* pdata) {
André Susano Pinto02a57972016-07-22 13:30:05 +0200436 if (IsQuitting()) {
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000437 return;
Taylor Brandstetter2b3bf6b2016-05-19 14:57:31 -0700438 }
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000439
440 // Keep thread safe
441 // Add to the priority queue. Gets sorted soonest first.
442 // Signal for the multiplexer to return.
443
jbauch9ccedc32016-02-25 01:14:56 -0800444 {
445 CritScope cs(&crit_);
446 Message msg;
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700447 msg.posted_from = posted_from;
jbauch9ccedc32016-02-25 01:14:56 -0800448 msg.phandler = phandler;
449 msg.message_id = id;
450 msg.pdata = pdata;
451 DelayedMessage dmsg(cmsDelay, tstamp, dmsgq_next_num_, msg);
452 dmsgq_.push(dmsg);
453 // If this message queue processes 1 message every millisecond for 50 days,
454 // we will wrap this number. Even then, only messages with identical times
455 // will be misordered, and then only briefly. This is probably ok.
nisse7ce109a2017-01-31 00:57:56 -0800456 ++dmsgq_next_num_;
457 RTC_DCHECK_NE(0, dmsgq_next_num_);
jbauch9ccedc32016-02-25 01:14:56 -0800458 }
459 WakeUpSocketServer();
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000460}
461
462int MessageQueue::GetDelay() {
463 CritScope cs(&crit_);
464
465 if (!msgq_.empty())
466 return 0;
467
468 if (!dmsgq_.empty()) {
469 int delay = TimeUntil(dmsgq_.top().msTrigger_);
470 if (delay < 0)
471 delay = 0;
472 return delay;
473 }
474
475 return kForever;
476}
477
Peter Boström0c4e06b2015-10-07 12:23:21 +0200478void MessageQueue::Clear(MessageHandler* phandler,
479 uint32_t id,
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000480 MessageList* removed) {
481 CritScope cs(&crit_);
482
483 // Remove messages with phandler
484
485 if (fPeekKeep_ && msgPeek_.Match(phandler, id)) {
486 if (removed) {
487 removed->push_back(msgPeek_);
488 } else {
489 delete msgPeek_.pdata;
490 }
491 fPeekKeep_ = false;
492 }
493
494 // Remove from ordered message queue
495
496 for (MessageList::iterator it = msgq_.begin(); it != msgq_.end();) {
497 if (it->Match(phandler, id)) {
498 if (removed) {
499 removed->push_back(*it);
500 } else {
501 delete it->pdata;
502 }
503 it = msgq_.erase(it);
504 } else {
505 ++it;
506 }
507 }
508
509 // Remove from priority queue. Not directly iterable, so use this approach
decurtis@webrtc.org2af30572015-02-21 01:59:50 +0000510
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000511 PriorityQueue::container_type::iterator new_end = dmsgq_.container().begin();
512 for (PriorityQueue::container_type::iterator it = new_end;
513 it != dmsgq_.container().end(); ++it) {
514 if (it->msg_.Match(phandler, id)) {
515 if (removed) {
516 removed->push_back(it->msg_);
517 } else {
518 delete it->msg_.pdata;
519 }
520 } else {
decurtis@webrtc.org2af30572015-02-21 01:59:50 +0000521 *new_end++ = *it;
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000522 }
523 }
524 dmsgq_.container().erase(new_end, dmsgq_.container().end());
525 dmsgq_.reheap();
526}
527
528void MessageQueue::Dispatch(Message *pmsg) {
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700529 TRACE_EVENT2("webrtc", "MessageQueue::Dispatch", "src_file_and_line",
530 pmsg->posted_from.file_and_line(), "src_func",
531 pmsg->posted_from.function_name());
532 int64_t start_time = TimeMillis();
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000533 pmsg->phandler->OnMessage(pmsg);
Taylor Brandstetter5d97a9a2016-06-10 14:17:27 -0700534 int64_t end_time = TimeMillis();
535 int64_t diff = TimeDiff(end_time, start_time);
536 if (diff >= kSlowDispatchLoggingThreshold) {
537 LOG(LS_INFO) << "Message took " << diff << "ms to dispatch. Posted from: "
538 << pmsg->posted_from.ToString();
539 }
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000540}
541
henrike@webrtc.orgf0488722014-05-13 18:00:26 +0000542} // namespace rtc