Fixing a couple cases that cause ProcessAllMessageQueues to hang.
The two situations are:
1. A thread is in the process of shutting down, so it won't handle any
more messages.
2. A message queue is cleared before it has a chance to process pending
messages.
In both of those cases, we should consider processing done at that
point.
R=honghaiz@webrtc.org, pthatcher@webrtc.org
Review URL: https://codereview.webrtc.org/2319303004 .
Cr-Commit-Position: refs/heads/master@{#14245}
diff --git a/webrtc/base/messagequeue.cc b/webrtc/base/messagequeue.cc
index ebf98f5..503d5af 100644
--- a/webrtc/base/messagequeue.cc
+++ b/webrtc/base/messagequeue.cc
@@ -126,17 +126,34 @@
}
void MessageQueueManager::ProcessAllMessageQueuesInternal() {
- // 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);
+ // This works by posting a delayed message at the current time and waiting
+ // 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 = 0;
+
+ // This class is used so that whether the posted message is processed, or the
+ // message queue is simply cleared, queues_not_done gets decremented.
+ class ScopedIncrement : public MessageData {
+ public:
+ ScopedIncrement(volatile int* value) : value_(value) {
+ AtomicOps::Increment(value_);
+ }
+ ~ScopedIncrement() override { AtomicOps::Decrement(value_); }
+
+ private:
+ volatile int* value_;
+ };
+
{
DebugNonReentrantCritScope cs(&crit_, &locked_);
- queues_not_done = static_cast<int>(message_queues_.size());
for (MessageQueue* queue : message_queues_) {
- queue->PostDelayed(RTC_FROM_HERE, 0, &handler);
+ if (queue->IsQuitting()) {
+ // If the queue is quitting, it's done processing messages so it can
+ // be ignored. If we tried to post a message to it, it would be dropped.
+ continue;
+ }
+ queue->PostDelayed(RTC_FROM_HERE, 0, nullptr, MQID_DISPOSE,
+ new ScopedIncrement(&queues_not_done));
}
}
// Note: One of the message queues may have been on this thread, which is why