OLD | NEW |
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" | 12 #include "webrtc/base/atomicops.h" |
13 #include "webrtc/base/checks.h" | 13 #include "webrtc/base/checks.h" |
14 #include "webrtc/base/common.h" | 14 #include "webrtc/base/common.h" |
15 #include "webrtc/base/logging.h" | 15 #include "webrtc/base/logging.h" |
16 #include "webrtc/base/messagequeue.h" | 16 #include "webrtc/base/messagequeue.h" |
| 17 #include "webrtc/base/stringencode.h" |
17 #include "webrtc/base/thread.h" | 18 #include "webrtc/base/thread.h" |
18 #include "webrtc/base/trace_event.h" | 19 #include "webrtc/base/trace_event.h" |
19 | 20 |
20 namespace rtc { | 21 namespace rtc { |
21 | 22 |
22 const int kMaxMsgLatency = 150; // 150 ms | 23 const int kMaxMsgLatency = 150; // 150 ms |
| 24 const int kSlowDispatchLoggingThreshold = 50; // 50 ms |
23 | 25 |
24 //------------------------------------------------------------------ | 26 //------------------------------------------------------------------ |
25 // MessageQueueManager | 27 // MessageQueueManager |
26 | 28 |
27 MessageQueueManager* MessageQueueManager::instance_ = NULL; | 29 MessageQueueManager* MessageQueueManager::instance_ = NULL; |
28 | 30 |
29 MessageQueueManager* MessageQueueManager::Instance() { | 31 MessageQueueManager* MessageQueueManager::Instance() { |
30 // Note: This is not thread safe, but it is first called before threads are | 32 // Note: This is not thread safe, but it is first called before threads are |
31 // spawned. | 33 // spawned. |
32 if (!instance_) | 34 if (!instance_) |
(...skipping 85 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
118 // Post a delayed message at the current time and wait for it to be dispatched | 120 // Post a delayed message at the current time and wait for it to be dispatched |
119 // on all queues, which will ensure that all messages that came before it were | 121 // on all queues, which will ensure that all messages that came before it were |
120 // also dispatched. | 122 // also dispatched. |
121 volatile int queues_not_done; | 123 volatile int queues_not_done; |
122 auto functor = [&queues_not_done] { AtomicOps::Decrement(&queues_not_done); }; | 124 auto functor = [&queues_not_done] { AtomicOps::Decrement(&queues_not_done); }; |
123 FunctorMessageHandler<void, decltype(functor)> handler(functor); | 125 FunctorMessageHandler<void, decltype(functor)> handler(functor); |
124 { | 126 { |
125 CritScope cs(&crit_); | 127 CritScope cs(&crit_); |
126 queues_not_done = static_cast<int>(message_queues_.size()); | 128 queues_not_done = static_cast<int>(message_queues_.size()); |
127 for (MessageQueue* queue : message_queues_) { | 129 for (MessageQueue* queue : message_queues_) { |
128 queue->PostDelayed(0, &handler); | 130 queue->PostDelayed(RTC_FROM_HERE, 0, &handler); |
129 } | 131 } |
130 } | 132 } |
131 // Note: One of the message queues may have been on this thread, which is why | 133 // 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 | 134 // we can't synchronously wait for queues_not_done to go to 0; we need to |
133 // process messages as well. | 135 // process messages as well. |
134 while (AtomicOps::AcquireLoad(&queues_not_done) > 0) { | 136 while (AtomicOps::AcquireLoad(&queues_not_done) > 0) { |
135 rtc::Thread::Current()->ProcessMessages(0); | 137 rtc::Thread::Current()->ProcessMessages(0); |
136 } | 138 } |
137 } | 139 } |
138 | 140 |
(...skipping 195 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
334 if (cmsElapsed >= cmsWait) | 336 if (cmsElapsed >= cmsWait) |
335 return false; | 337 return false; |
336 } | 338 } |
337 } | 339 } |
338 return false; | 340 return false; |
339 } | 341 } |
340 | 342 |
341 void MessageQueue::ReceiveSends() { | 343 void MessageQueue::ReceiveSends() { |
342 } | 344 } |
343 | 345 |
344 void MessageQueue::Post(MessageHandler* phandler, | 346 void MessageQueue::Post(const Location& posted_from, |
| 347 MessageHandler* phandler, |
345 uint32_t id, | 348 uint32_t id, |
346 MessageData* pdata, | 349 MessageData* pdata, |
347 bool time_sensitive) { | 350 bool time_sensitive) { |
348 if (fStop_) | 351 if (fStop_) |
349 return; | 352 return; |
350 | 353 |
351 // Keep thread safe | 354 // Keep thread safe |
352 // Add the message to the end of the queue | 355 // Add the message to the end of the queue |
353 // Signal for the multiplexer to return | 356 // Signal for the multiplexer to return |
354 | 357 |
355 { | 358 { |
356 CritScope cs(&crit_); | 359 CritScope cs(&crit_); |
357 Message msg; | 360 Message msg; |
| 361 msg.posted_from = posted_from; |
358 msg.phandler = phandler; | 362 msg.phandler = phandler; |
359 msg.message_id = id; | 363 msg.message_id = id; |
360 msg.pdata = pdata; | 364 msg.pdata = pdata; |
361 if (time_sensitive) { | 365 if (time_sensitive) { |
362 msg.ts_sensitive = TimeMillis() + kMaxMsgLatency; | 366 msg.ts_sensitive = TimeMillis() + kMaxMsgLatency; |
363 } | 367 } |
364 msgq_.push_back(msg); | 368 msgq_.push_back(msg); |
365 } | 369 } |
366 WakeUpSocketServer(); | 370 WakeUpSocketServer(); |
367 } | 371 } |
368 | 372 |
369 void MessageQueue::PostDelayed(int cmsDelay, | 373 void MessageQueue::PostDelayed(const Location& posted_from, |
| 374 int cmsDelay, |
370 MessageHandler* phandler, | 375 MessageHandler* phandler, |
371 uint32_t id, | 376 uint32_t id, |
372 MessageData* pdata) { | 377 MessageData* pdata) { |
373 return DoDelayPost(cmsDelay, TimeAfter(cmsDelay), phandler, id, pdata); | 378 return DoDelayPost(posted_from, cmsDelay, TimeAfter(cmsDelay), phandler, id, |
| 379 pdata); |
374 } | 380 } |
375 | 381 |
376 void MessageQueue::PostAt(uint32_t tstamp, | 382 void MessageQueue::PostAt(const Location& posted_from, |
| 383 uint32_t tstamp, |
377 MessageHandler* phandler, | 384 MessageHandler* phandler, |
378 uint32_t id, | 385 uint32_t id, |
379 MessageData* pdata) { | 386 MessageData* pdata) { |
380 // This should work even if it is used (unexpectedly). | 387 // This should work even if it is used (unexpectedly). |
381 int64_t delay = static_cast<uint32_t>(TimeMillis()) - tstamp; | 388 int64_t delay = static_cast<uint32_t>(TimeMillis()) - tstamp; |
382 return DoDelayPost(delay, tstamp, phandler, id, pdata); | 389 return DoDelayPost(posted_from, delay, tstamp, phandler, id, pdata); |
383 } | 390 } |
384 | 391 |
385 void MessageQueue::PostAt(int64_t tstamp, | 392 void MessageQueue::PostAt(const Location& posted_from, |
| 393 int64_t tstamp, |
386 MessageHandler* phandler, | 394 MessageHandler* phandler, |
387 uint32_t id, | 395 uint32_t id, |
388 MessageData* pdata) { | 396 MessageData* pdata) { |
389 return DoDelayPost(TimeUntil(tstamp), tstamp, phandler, id, pdata); | 397 return DoDelayPost(posted_from, TimeUntil(tstamp), tstamp, phandler, id, |
| 398 pdata); |
390 } | 399 } |
391 | 400 |
392 void MessageQueue::DoDelayPost(int64_t cmsDelay, | 401 void MessageQueue::DoDelayPost(const Location& posted_from, |
| 402 int64_t cmsDelay, |
393 int64_t tstamp, | 403 int64_t tstamp, |
394 MessageHandler* phandler, | 404 MessageHandler* phandler, |
395 uint32_t id, | 405 uint32_t id, |
396 MessageData* pdata) { | 406 MessageData* pdata) { |
397 if (fStop_) { | 407 if (fStop_) { |
398 return; | 408 return; |
399 } | 409 } |
400 | 410 |
401 // Keep thread safe | 411 // Keep thread safe |
402 // Add to the priority queue. Gets sorted soonest first. | 412 // Add to the priority queue. Gets sorted soonest first. |
403 // Signal for the multiplexer to return. | 413 // Signal for the multiplexer to return. |
404 | 414 |
405 { | 415 { |
406 CritScope cs(&crit_); | 416 CritScope cs(&crit_); |
407 Message msg; | 417 Message msg; |
| 418 msg.posted_from = posted_from; |
408 msg.phandler = phandler; | 419 msg.phandler = phandler; |
409 msg.message_id = id; | 420 msg.message_id = id; |
410 msg.pdata = pdata; | 421 msg.pdata = pdata; |
411 DelayedMessage dmsg(cmsDelay, tstamp, dmsgq_next_num_, msg); | 422 DelayedMessage dmsg(cmsDelay, tstamp, dmsgq_next_num_, msg); |
412 dmsgq_.push(dmsg); | 423 dmsgq_.push(dmsg); |
413 // If this message queue processes 1 message every millisecond for 50 days, | 424 // If this message queue processes 1 message every millisecond for 50 days, |
414 // we will wrap this number. Even then, only messages with identical times | 425 // we will wrap this number. Even then, only messages with identical times |
415 // will be misordered, and then only briefly. This is probably ok. | 426 // will be misordered, and then only briefly. This is probably ok. |
416 VERIFY(0 != ++dmsgq_next_num_); | 427 VERIFY(0 != ++dmsgq_next_num_); |
417 } | 428 } |
(...skipping 60 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
478 } | 489 } |
479 } else { | 490 } else { |
480 *new_end++ = *it; | 491 *new_end++ = *it; |
481 } | 492 } |
482 } | 493 } |
483 dmsgq_.container().erase(new_end, dmsgq_.container().end()); | 494 dmsgq_.container().erase(new_end, dmsgq_.container().end()); |
484 dmsgq_.reheap(); | 495 dmsgq_.reheap(); |
485 } | 496 } |
486 | 497 |
487 void MessageQueue::Dispatch(Message *pmsg) { | 498 void MessageQueue::Dispatch(Message *pmsg) { |
488 TRACE_EVENT0("webrtc", "MessageQueue::Dispatch"); | 499 TRACE_EVENT2("webrtc", "MessageQueue::Dispatch", "src_file_and_line", |
| 500 pmsg->posted_from.file_and_line(), "src_func", |
| 501 pmsg->posted_from.function_name()); |
| 502 int64_t start_time = TimeMillis(); |
489 pmsg->phandler->OnMessage(pmsg); | 503 pmsg->phandler->OnMessage(pmsg); |
| 504 int64_t end_time = TimeMillis(); |
| 505 int64_t diff = TimeDiff(end_time, start_time); |
| 506 if (diff >= kSlowDispatchLoggingThreshold) { |
| 507 LOG(LS_INFO) << "Message took " << diff << "ms to dispatch. Posted from: " |
| 508 << pmsg->posted_from.ToString(); |
| 509 } |
490 } | 510 } |
491 | 511 |
492 } // namespace rtc | 512 } // namespace rtc |
OLD | NEW |