Reimplement ReceivedSyncMsgQueue::DispatchMessages

ipc/ipc_sync_channel.cc

 // Copyright (c) 2011 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "ipc/ipc_sync_channel.h"
 
 #include "base/bind.h"
 #include "base/lazy_instance.h"
 #include "base/location.h"
 #include "base/logging.h"
@@ skipping 46 matching lines @@
     bool was_task_pending;
     {
       base::AutoLock auto_lock(message_lock_);
 
       was_task_pending = task_pending_;
       task_pending_ = true;
 
       // We set the event in case the listener thread is blocked (or is about
       // to). In case it's not, the PostTask dispatches the messages.
       message_queue_.push_back(QueuedMessage(new Message(msg), context));
+      message_queue_version_++;
     }
 
     dispatch_event_.Signal();
     if (!was_task_pending) {
       listener_message_loop_->PostTask(
           FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchMessagesTask,
                                 this, scoped_refptr<SyncContext>(context)));
     }
   }
 
   void QueueReply(const Message &msg, SyncChannel::SyncContext* context) {
     received_replies_.push_back(QueuedMessage(new Message(msg), context));
   }
 
   // Called on the listener's thread to process any queues synchronous
   // messages.
   void DispatchMessagesTask(SyncContext* context) {
     {
       base::AutoLock auto_lock(message_lock_);
       task_pending_ = false;
     }
     context->DispatchMessages();
   }
 
   void DispatchMessages(SyncContext* dispatching_context) {
-    SyncMessageQueue delayed_queue;
+    bool first_time = true;
+    uint32 expected_version = 0;
+    SyncMessageQueue::iterator it;
     while (true) {
-      Message* message;
+      Message* message = NULL;
       scoped_refptr<SyncChannel::SyncContext> context;
       {
         base::AutoLock auto_lock(message_lock_);
-        if (message_queue_.empty()) {
-          message_queue_ = delayed_queue;
-          break;
+        if (first_time || message_queue_version_ != expected_version) {
+          it = message_queue_.begin();
+          first_time = false;
         }
+        for (; it != message_queue_.end(); it++) {
+          if (!it->context->restrict_dispatch() ||
+              it->context == dispatching_context) {
+            message = it->message;
+            context = it->context;
+            it = message_queue_.erase(it);
+            message_queue_version_++;
+            expected_version = message_queue_version_;
+            break;
+          }
+        }
+      }
 
-        message = message_queue_.front().message;
-        context = message_queue_.front().context;
-        message_queue_.pop_front();
-      }
-      if (context->restrict_dispatch() && context != dispatching_context) {
-        delayed_queue.push_back(QueuedMessage(message, context));
-      } else {
-        context->OnDispatchMessage(*message);
-        delete message;
-      }
+      if (message == NULL)
+        break;
+      context->OnDispatchMessage(*message);
+      delete message;
     }
   }
 
   // SyncChannel calls this in its destructor.
   void RemoveContext(SyncContext* context) {
     base::AutoLock auto_lock(message_lock_);
 
     SyncMessageQueue::iterator iter = message_queue_.begin();
     while (iter != message_queue_.end()) {
       if (iter->context == context) {
         delete iter->message;
         iter = message_queue_.erase(iter);
+        message_queue_version_++;
       } else {
         iter++;
       }
     }
 
     if (--listener_count_ == 0) {
       DCHECK(lazy_tls_ptr_.Pointer()->Get());
       lazy_tls_ptr_.Pointer()->Set(NULL);
     }
   }
@@ skipping 27 matching lines @@
   void set_top_send_done_watcher(base::WaitableEventWatcher* watcher) {
     top_send_done_watcher_ = watcher;
   }
 
  private:
   friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>;
 
   // See the comment in SyncChannel::SyncChannel for why this event is created
   // as manual reset.
   ReceivedSyncMsgQueue() :
+      message_queue_version_(0),
       dispatch_event_(true, false),
       listener_message_loop_(base::MessageLoopProxy::current()),
       task_pending_(false),
       listener_count_(0),
       top_send_done_watcher_(NULL) {
   }
 
   ~ReceivedSyncMsgQueue() {}
 
   // Holds information about a queued synchronous message or reply.
   struct QueuedMessage {
     QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
     Message* message;
     scoped_refptr<SyncChannel::SyncContext> context;
   };
 
-  typedef std::deque<QueuedMessage> SyncMessageQueue;
+  typedef std::list<QueuedMessage> SyncMessageQueue;
   SyncMessageQueue message_queue_;
+  uint32 message_queue_version_;  // Used to signal DispatchMessages to rescan
 
   std::vector<QueuedMessage> received_replies_;
 
   // Set when we got a synchronous message that we must respond to as the
   // sender needs its reply before it can reply to our original synchronous
   // message.
   WaitableEvent dispatch_event_;
   scoped_refptr<base::MessageLoopProxy> listener_message_loop_;
   base::Lock message_lock_;
   bool task_pending_;
@@ skipping 319 matching lines @@
   // Ideally we only want to watch this object when running a nested message
   // loop.  However, we don't know when it exits if there's another nested
   // message loop running under it or not, so we wouldn't know whether to
   // stop or keep watching.  So we always watch it, and create the event as
   // manual reset since the object watcher might otherwise reset the event
   // when we're doing a WaitMany.
   dispatch_watcher_.StartWatching(sync_context()->GetDispatchEvent(), this);
 }
 
 }  // namespace IPC