Use MessageLoopProxy instead of MessageLoop to dispatch notifications in ObserverListThreadsafe.

base/observer_list_threadsafe.h

 // 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.
 
 #ifndef BASE_OBSERVER_LIST_THREADSAFE_H_
 #define BASE_OBSERVER_LIST_THREADSAFE_H_
 #pragma once
 
 #include <algorithm>
 #include <map>
 
 #include "base/basictypes.h"
 #include "base/callback_old.h"
 #include "base/logging.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop.h"
+#include "base/message_loop_proxy.h"
 #include "base/observer_list.h"
 #include "base/task.h"
 
 ///////////////////////////////////////////////////////////////////////////////
 //
 // OVERVIEW:
 //
 //   A thread-safe container for a list of observers.
 //   This is similar to the observer_list (see observer_list.h), but it
 //   is more robust for multi-threaded situations.
@@ skipping 52 matching lines @@
   typedef typename ObserverList<ObserverType>::NotificationType
       NotificationType;
 
   ObserverListThreadSafe()
       : type_(ObserverListBase<ObserverType>::NOTIFY_ALL) {}
   explicit ObserverListThreadSafe(NotificationType type) : type_(type) {}
 
   // Add an observer to the list.  An observer should not be added to
   // the same list more than once.
   void AddObserver(ObserverType* obs) {
-    ObserverList<ObserverType>* list = NULL;
-    MessageLoop* loop = MessageLoop::current();
     // TODO(mbelshe): Get rid of this check.  Its needed right now because
     //                Time currently triggers usage of the ObserverList.
     //                And unittests use time without a MessageLoop.
-    if (!loop)
+    if (!MessageLoop::current())
       return;  // Some unittests may access this without a message loop.
+    base::MessageLoopProxy* loop_proxy = GetOrCreateMessageLoopProxy();
+    ObserverList<ObserverType>* list = NULL;
     {
       base::AutoLock lock(list_lock_);
-      if (observer_lists_.find(loop) == observer_lists_.end())
-        observer_lists_[loop] = new ObserverList<ObserverType>(type_);
-      list = observer_lists_[loop];
+      if (observer_lists_.find(loop_proxy) == observer_lists_.end())
+        observer_lists_[loop_proxy] = new ObserverList<ObserverType>(type_);
+      list = observer_lists_[loop_proxy];
     }
     list->AddObserver(obs);
   }
 
   // Remove an observer from the list if it is in the list.
   // If there are pending notifications in-transit to the observer, they will
   // be aborted.
   // If the observer to be removed is in the list, RemoveObserver MUST
   // be called from the same thread which called AddObserver.
   void RemoveObserver(ObserverType* obs) {
+    if (!MessageLoop::current())
+      return;  // On shutdown, it is possible that current() is already null.
+    base::MessageLoopProxy* loop_proxy = GetExistingMessageLoopProxy();
+    if (!loop_proxy) {
+      // This will happen if we try to remove an observer on a thread
+      // we never added an observer for.
+      return;
+    }
     ObserverList<ObserverType>* list = NULL;
-    MessageLoop* loop = MessageLoop::current();
-    if (!loop)
-      return;  // On shutdown, it is possible that current() is already null.
     {
       base::AutoLock lock(list_lock_);
-      typename ObserversListMap::iterator it = observer_lists_.find(loop);
+      typename ObserversListMap::iterator it = observer_lists_.find(loop_proxy);
       if (it == observer_lists_.end()) {
-        // This may happen if we try to remove an observer on a thread
-        // we never added an observer for.
+        // This can happen if a previously-used MessageLoopProxy
+        // no longer has any observers (we don't clean out the cache).
         return;
       }
       list = it->second;
 
       // If we're about to remove the last observer from the list,
       // then we can remove this observer_list entirely.
       if (list->HasObserver(obs) && list->size() == 1)
         observer_lists_.erase(it);
     }
     list->RemoveObserver(obs);
@@ skipping 54 matching lines @@
     for (it = observer_lists_.begin(); it != observer_lists_.end(); ++it)
       delete (*it).second;
     observer_lists_.clear();
   }
 
   template <class Method, class Params>
   void Notify(const UnboundMethod<ObserverType, Method, Params>& method) {
     base::AutoLock lock(list_lock_);
     typename ObserversListMap::iterator it;
     for (it = observer_lists_.begin(); it != observer_lists_.end(); ++it) {
-      MessageLoop* loop = (*it).first;
+      scoped_refptr<base::MessageLoopProxy> loop = (*it).first;
       ObserverList<ObserverType>* list = (*it).second;
       loop->PostTask(
           FROM_HERE,
           NewRunnableMethod(this,
               &ObserverListThreadSafe<ObserverType>::
                  template NotifyWrapper<Method, Params>, list, method));
     }
   }
 
   // Wrapper which is called to fire the notifications for each thread's
   // ObserverList.  This function MUST be called on the thread which owns
   // the unsafe ObserverList.
   template <class Method, class Params>
   void NotifyWrapper(ObserverList<ObserverType>* list,
       const UnboundMethod<ObserverType, Method, Params>& method) {
 
     // Check that this list still needs notifications.
     {
       base::AutoLock lock(list_lock_);
       typename ObserversListMap::iterator it =
-          observer_lists_.find(MessageLoop::current());
+          observer_lists_.find(GetExistingMessageLoopProxy());
 
       // The ObserverList could have been removed already.  In fact, it could
       // have been removed and then re-added!  If the master list's loop
       // does not match this one, then we do not need to finish this
       // notification.
       if (it == observer_lists_.end() || it->second != list)
         return;
     }
 
     {
       typename ObserverList<ObserverType>::Iterator it(*list);
       ObserverType* obs;
       while ((obs = it.GetNext()) != NULL)
         method.Run(obs);
     }
 
     // If there are no more observers on the list, we can now delete it.
     if (list->size() == 0) {
       {
         base::AutoLock lock(list_lock_);
         // Remove |list| if it's not already removed.
         // This can happen if multiple observers got removed in a notification.
         // See http://crbug.com/55725.
         typename ObserversListMap::iterator it =
-            observer_lists_.find(MessageLoop::current());
+            observer_lists_.find(GetExistingMessageLoopProxy());
         if (it != observer_lists_.end() && it->second == list)
           observer_lists_.erase(it);
       }
       delete list;
     }
   }
 
-  typedef std::map<MessageLoop*, ObserverList<ObserverType>*> ObserversListMap;
+  base::MessageLoopProxy* GetExistingMessageLoopProxy() {
+    base::AutoLock lock(message_loop_lock_);
+    return GetExistingMessageLoopProxyUnlocked();
+  }
 
-  // These are marked mutable to facilitate having NotifyAll be const.
+  base::MessageLoopProxy* GetOrCreateMessageLoopProxy() {
+    base::AutoLock lock(message_loop_lock_);
+    base::MessageLoopProxy* loop_proxy = GetExistingMessageLoopProxyUnlocked();
+    if (!loop_proxy) {
+      scoped_refptr<base::MessageLoopProxy> loop_proxy_refptr =
+          base::MessageLoopProxy::CreateForCurrentThread();
+      message_loop_proxy_cache_[MessageLoop::current()] = loop_proxy_refptr;
+      loop_proxy = loop_proxy_refptr.get();
+    }
+    return loop_proxy;
+  }
+
+  base::MessageLoopProxy* GetExistingMessageLoopProxyUnlocked() {

[darin (slow to review), 2011/08/08 18:18:26]

The "Unlocked" suffix is a little confusing to me.  It sounds like the function
is called when we do not have the lock held, but that's actually the opposite of
how it should be used.  We often use a suffix of "_Locked" to indicate when a
method should be called with the lock already held.

+    typename MessageLoopMap::iterator it =
+        message_loop_proxy_cache_.find(MessageLoop::current());
+    if (it != message_loop_proxy_cache_.end())
+      return (it->second).get();
+    return NULL;
+  }
+
+  typedef std::map<base::MessageLoopProxy*, ObserverList<ObserverType>*>
+      ObserversListMap;
+  typedef std::map<MessageLoop*, scoped_refptr<base::MessageLoopProxy> >
+      MessageLoopMap;
+
   base::Lock list_lock_;  // Protects the observer_lists_.
   ObserversListMap observer_lists_;
+  base::Lock message_loop_lock_;  // Protects message_loop_proxy_cache_;
+  MessageLoopMap message_loop_proxy_cache_;
   const NotificationType type_;
 
   DISALLOW_COPY_AND_ASSIGN(ObserverListThreadSafe);
 };
 
 #endif  // BASE_OBSERVER_LIST_THREADSAFE_H_