DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (d1ed7de67f5a)

VCS Links

Line Code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203
// Copyright (c) 2006-2008 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_

#include <vector>
#include <algorithm>

#include "base/basictypes.h"
#include "base/logging.h"
#include "base/message_loop.h"
#include "base/observer_list.h"
#include "base/ref_counted.h"
#include "base/task.h"

namespace base {

///////////////////////////////////////////////////////////////////////////////
//
// 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.
//
//   The following use cases are supported:
//    * Observers can register for notifications from any thread.
//      Callbacks to the observer will occur on the same thread where
//      the observer initially called AddObserver() from.
//    * Any thread may trigger a notification via NOTIFY_OBSERVERS.
//    * Observers can remove themselves from the observer list inside
//      of a callback.
//    * If one thread is notifying observers concurrently with an observer
//      removing itself from the observer list, the notifications will
//      be silently dropped.
//
//   The drawback of the threadsafe observer list is that notifications
//   are not as real-time as the non-threadsafe version of this class.
//   Notifications will always be done via PostTask() to another thread,
//   whereas with the non-thread-safe observer_list, notifications happen
//   synchronously and immediately.
//
//   IMPLEMENTATION NOTES
//   The ObserverListThreadSafe maintains an ObserverList for each thread
//   which uses the ThreadSafeObserver.  When Notifying the observers,
//   we simply call PostTask to each registered thread, and then each thread
//   will notify its regular ObserverList.
//
///////////////////////////////////////////////////////////////////////////////
template <class ObserverType>
class ObserverListThreadSafe
    : public base::RefCountedThreadSafe<ObserverListThreadSafe<ObserverType> > {
 public:
  ObserverListThreadSafe() {}

  ~ObserverListThreadSafe() {
    typename ObserversListMap::const_iterator it;
    for (it = observer_lists_.begin(); it != observer_lists_.end(); ++it)
      delete (*it).second;
    observer_lists_.clear();
  }

  // Add an observer to the list.
  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)
      return;  // Some unittests may access this without a message loop.
    {
      AutoLock lock(list_lock_);
      if (observer_lists_.find(loop) == observer_lists_.end())
        observer_lists_[loop] = new ObserverList<ObserverType>();
      list = observer_lists_[loop];
    }
    list->AddObserver(obs);
  }

  // Remove an observer from the list.
  // If there are pending notifications in-transit to the observer, they will
  // be aborted.
  // RemoveObserver MUST be called from the same thread which called
  // AddObserver.
  void RemoveObserver(ObserverType* obs) {
    ObserverList<ObserverType>* list = NULL;
    MessageLoop* loop = MessageLoop::current();
    if (!loop)
      return;  // On shutdown, it is possible that current() is already null.
    {
      AutoLock lock(list_lock_);
      list = observer_lists_[loop];
      if (!list) {
        NOTREACHED() << "RemoveObserver called on for unknown thread";
        return;
      }

      // If we're about to remove the last observer from the list,
      // then we can remove this observer_list entirely.
      if (list->size() == 1)
        observer_lists_.erase(loop);
    }
    list->RemoveObserver(obs);

    // If RemoveObserver is called from a notification, the size will be
    // nonzero.  Instead of deleting here, the NotifyWrapper will delete
    // when it finishes iterating.
    if (list->size() == 0)
      delete list;
  }

  // Notify methods.
  // Make a thread-safe callback to each Observer in the list.
  // Note, these calls are effectively asynchronous.  You cannot assume
  // that at the completion of the Notify call that all Observers have
  // been Notified.  The notification may still be pending delivery.
  template <class Method>
  void Notify(Method m) {
    UnboundMethod<ObserverType, Method, Tuple0> method(m, MakeTuple());
    Notify<Method, Tuple0>(method);
  }

  template <class Method, class A>
  void Notify(Method m, const A &a) {
    UnboundMethod<ObserverType, Method, Tuple1<A> > method(m, MakeTuple(a));
    Notify<Method, Tuple1<A> >(method);
  }

  // TODO(mbelshe):  Add more wrappers for Notify() with more arguments.

 private:
  template <class Method, class Params>
  void Notify(const UnboundMethod<ObserverType, Method, Params>& method) {
    AutoLock lock(list_lock_);
    typename ObserversListMap::iterator it;
    for (it = observer_lists_.begin(); it != observer_lists_.end(); ++it) {
      MessageLoop* 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.
    {
      AutoLock lock(list_lock_);
      typename ObserversListMap::iterator it =
          observer_lists_.find(MessageLoop::current());

      // 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) {
#ifndef NDEBUG
      {
        AutoLock lock(list_lock_);
        // Verify this list is no longer registered.
        typename ObserversListMap::iterator it =
            observer_lists_.find(MessageLoop::current());
        DCHECK(it == observer_lists_.end() || it->second != list);
      }
#endif
      delete list;
    }
  }

  typedef std::map<MessageLoop*, ObserverList<ObserverType>*> ObserversListMap;

  // These are marked mutable to facilitate having NotifyAll be const.
  Lock list_lock_;  // Protects the observer_lists_.
  ObserversListMap observer_lists_;

  DISALLOW_EVIL_CONSTRUCTORS(ObserverListThreadSafe);
};

} // namespace base

#endif  // BASE_OBSERVER_LIST_THREADSAFE_H_