observer_list_threadsafe.h 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280
  1. // Copyright (c) 2012 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #ifndef BASE_OBSERVER_LIST_THREADSAFE_H_
  5. #define BASE_OBSERVER_LIST_THREADSAFE_H_
  6. #include <unordered_map>
  7. #include <utility>
  8. #include "base/base_export.h"
  9. #include "base/bind.h"
  10. #include "base/check.h"
  11. #include "base/check_op.h"
  12. #include "base/containers/contains.h"
  13. #include "base/dcheck_is_on.h"
  14. #include "base/lazy_instance.h"
  15. #include "base/location.h"
  16. #include "base/memory/raw_ptr.h"
  17. #include "base/memory/ref_counted.h"
  18. #include "base/observer_list.h"
  19. #include "base/synchronization/lock.h"
  20. #include "base/task/sequenced_task_runner.h"
  21. #include "base/threading/sequenced_task_runner_handle.h"
  22. #include "base/threading/thread_local.h"
  23. #include "build/build_config.h"
  24. ///////////////////////////////////////////////////////////////////////////////
  25. //
  26. // OVERVIEW:
  27. //
  28. // A thread-safe container for a list of observers. This is similar to the
  29. // observer_list (see observer_list.h), but it is more robust for multi-
  30. // threaded situations.
  31. //
  32. // The following use cases are supported:
  33. // * Observers can register for notifications from any sequence. They are
  34. // always notified on the sequence from which they were registered.
  35. // * Any sequence may trigger a notification via Notify().
  36. // * Observers can remove themselves from the observer list inside of a
  37. // callback.
  38. // * If one sequence is notifying observers concurrently with an observer
  39. // removing itself from the observer list, the notifications will be
  40. // silently dropped.
  41. //
  42. // The drawback of the threadsafe observer list is that notifications are not
  43. // as real-time as the non-threadsafe version of this class. Notifications
  44. // will always be done via PostTask() to another sequence, whereas with the
  45. // non-thread-safe ObserverList, notifications happen synchronously.
  46. //
  47. // Note: this class previously supported synchronous notifications for
  48. // same-sequence observers, but it was error-prone and removed in
  49. // crbug.com/1193750, think twice before re-considering this paradigm.
  50. //
  51. ///////////////////////////////////////////////////////////////////////////////
  52. namespace base {
  53. namespace internal {
  54. class BASE_EXPORT ObserverListThreadSafeBase
  55. : public RefCountedThreadSafe<ObserverListThreadSafeBase> {
  56. public:
  57. ObserverListThreadSafeBase() = default;
  58. ObserverListThreadSafeBase(const ObserverListThreadSafeBase&) = delete;
  59. ObserverListThreadSafeBase& operator=(const ObserverListThreadSafeBase&) =
  60. delete;
  61. protected:
  62. template <typename ObserverType, typename Method>
  63. struct Dispatcher;
  64. template <typename ObserverType, typename ReceiverType, typename... Params>
  65. struct Dispatcher<ObserverType, void (ReceiverType::*)(Params...)> {
  66. static void Run(void (ReceiverType::*m)(Params...),
  67. Params... params,
  68. ObserverType* obj) {
  69. (obj->*m)(std::forward<Params>(params)...);
  70. }
  71. };
  72. struct NotificationDataBase {
  73. NotificationDataBase(void* observer_list_in, const Location& from_here_in)
  74. : observer_list(observer_list_in), from_here(from_here_in) {}
  75. raw_ptr<void> observer_list;
  76. Location from_here;
  77. };
  78. virtual ~ObserverListThreadSafeBase() = default;
  79. static LazyInstance<ThreadLocalPointer<const NotificationDataBase>>::Leaky
  80. tls_current_notification_;
  81. private:
  82. friend class RefCountedThreadSafe<ObserverListThreadSafeBase>;
  83. };
  84. } // namespace internal
  85. template <class ObserverType>
  86. class ObserverListThreadSafe : public internal::ObserverListThreadSafeBase {
  87. public:
  88. enum class AddObserverResult {
  89. kBecameNonEmpty,
  90. kWasAlreadyNonEmpty,
  91. };
  92. enum class RemoveObserverResult {
  93. kWasOrBecameEmpty,
  94. kRemainsNonEmpty,
  95. };
  96. ObserverListThreadSafe() = default;
  97. explicit ObserverListThreadSafe(ObserverListPolicy policy)
  98. : policy_(policy) {}
  99. ObserverListThreadSafe(const ObserverListThreadSafe&) = delete;
  100. ObserverListThreadSafe& operator=(const ObserverListThreadSafe&) = delete;
  101. // Adds |observer| to the list. |observer| must not already be in the list.
  102. AddObserverResult AddObserver(ObserverType* observer) {
  103. DCHECK(SequencedTaskRunnerHandle::IsSet())
  104. << "An observer can only be registered when SequencedTaskRunnerHandle "
  105. "is set. If this is in a unit test, you're likely merely missing a "
  106. "base::test::(SingleThread)TaskEnvironment in your fixture. "
  107. "Otherwise, try running this code on a named thread (main/UI/IO) or "
  108. "from a task posted to a base::SequencedTaskRunner or "
  109. "base::SingleThreadTaskRunner.";
  110. AutoLock auto_lock(lock_);
  111. bool was_empty = observers_.empty();
  112. // Add |observer| to the list of observers.
  113. DCHECK(!Contains(observers_, observer));
  114. const scoped_refptr<SequencedTaskRunner> task_runner =
  115. SequencedTaskRunnerHandle::Get();
  116. // Each observer gets a unique identifier. These unique identifiers are used
  117. // to avoid execution of pending posted-tasks over removed or released
  118. // observers.
  119. const size_t observer_id = ++observer_id_counter_;
  120. ObserverTaskRunnerInfo task_info = {task_runner, observer_id};
  121. observers_[observer] = std::move(task_info);
  122. // If this is called while a notification is being dispatched on this thread
  123. // and |policy_| is ALL, |observer| must be notified (if a notification is
  124. // being dispatched on another thread in parallel, the notification may or
  125. // may not make it to |observer| depending on the outcome of the race to
  126. // |lock_|).
  127. if (policy_ == ObserverListPolicy::ALL) {
  128. const NotificationDataBase* current_notification =
  129. tls_current_notification_.Get().Get();
  130. if (current_notification && current_notification->observer_list == this) {
  131. const NotificationData* notification_data =
  132. static_cast<const NotificationData*>(current_notification);
  133. task_runner->PostTask(
  134. current_notification->from_here,
  135. BindOnce(&ObserverListThreadSafe<ObserverType>::NotifyWrapper, this,
  136. observer,
  137. NotificationData(this, observer_id,
  138. current_notification->from_here,
  139. notification_data->method)));
  140. }
  141. }
  142. return was_empty ? AddObserverResult::kBecameNonEmpty
  143. : AddObserverResult::kWasAlreadyNonEmpty;
  144. }
  145. // Remove an observer from the list if it is in the list.
  146. //
  147. // If a notification was sent to the observer but hasn't started to run yet,
  148. // it will be aborted. If a notification has started to run, removing the
  149. // observer won't stop it.
  150. RemoveObserverResult RemoveObserver(ObserverType* observer) {
  151. AutoLock auto_lock(lock_);
  152. observers_.erase(observer);
  153. return observers_.empty() ? RemoveObserverResult::kWasOrBecameEmpty
  154. : RemoveObserverResult::kRemainsNonEmpty;
  155. }
  156. // Verifies that the list is currently empty (i.e. there are no observers).
  157. void AssertEmpty() const {
  158. #if DCHECK_IS_ON()
  159. AutoLock auto_lock(lock_);
  160. DCHECK(observers_.empty());
  161. #endif
  162. }
  163. // Asynchronously invokes a callback on all observers, on their registration
  164. // sequence. You cannot assume that at the completion of the Notify call that
  165. // all Observers have been Notified. The notification may still be pending
  166. // delivery.
  167. template <typename Method, typename... Params>
  168. void Notify(const Location& from_here, Method m, Params&&... params) {
  169. RepeatingCallback<void(ObserverType*)> method =
  170. BindRepeating(&Dispatcher<ObserverType, Method>::Run, m,
  171. std::forward<Params>(params)...);
  172. AutoLock lock(lock_);
  173. for (const auto& observer : observers_) {
  174. observer.second.task_runner->PostTask(
  175. from_here,
  176. BindOnce(&ObserverListThreadSafe<ObserverType>::NotifyWrapper, this,
  177. observer.first,
  178. NotificationData(this, observer.second.observer_id,
  179. from_here, method)));
  180. }
  181. }
  182. private:
  183. friend class RefCountedThreadSafe<ObserverListThreadSafeBase>;
  184. struct NotificationData : public NotificationDataBase {
  185. NotificationData(ObserverListThreadSafe* observer_list_in,
  186. size_t observer_id_in,
  187. const Location& from_here_in,
  188. const RepeatingCallback<void(ObserverType*)>& method_in)
  189. : NotificationDataBase(observer_list_in, from_here_in),
  190. method(method_in),
  191. observer_id(observer_id_in) {}
  192. RepeatingCallback<void(ObserverType*)> method;
  193. size_t observer_id;
  194. };
  195. ~ObserverListThreadSafe() override = default;
  196. void NotifyWrapper(ObserverType* observer,
  197. const NotificationData& notification) {
  198. {
  199. AutoLock auto_lock(lock_);
  200. // Check whether the observer still needs a notification.
  201. DCHECK_EQ(notification.observer_list, this);
  202. auto it = observers_.find(observer);
  203. if (it == observers_.end() ||
  204. it->second.observer_id != notification.observer_id) {
  205. return;
  206. }
  207. DCHECK(it->second.task_runner->RunsTasksInCurrentSequence());
  208. }
  209. // Keep track of the notification being dispatched on the current thread.
  210. // This will be used if the callback below calls AddObserver().
  211. //
  212. // Note: |tls_current_notification_| may not be nullptr if this runs in a
  213. // nested loop started by a notification callback. In that case, it is
  214. // important to save the previous value to restore it later.
  215. auto& tls_current_notification = tls_current_notification_.Get();
  216. const NotificationDataBase* const previous_notification =
  217. tls_current_notification.Get();
  218. tls_current_notification.Set(&notification);
  219. // Invoke the callback.
  220. notification.method.Run(observer);
  221. // Reset the notification being dispatched on the current thread to its
  222. // previous value.
  223. tls_current_notification.Set(previous_notification);
  224. }
  225. const ObserverListPolicy policy_ = ObserverListPolicy::ALL;
  226. mutable Lock lock_;
  227. size_t observer_id_counter_ GUARDED_BY(lock_) = 0;
  228. struct ObserverTaskRunnerInfo {
  229. scoped_refptr<SequencedTaskRunner> task_runner;
  230. size_t observer_id = 0;
  231. };
  232. // Keys are observers. Values are the SequencedTaskRunners on which they must
  233. // be notified.
  234. std::unordered_map<ObserverType*, ObserverTaskRunnerInfo> observers_
  235. GUARDED_BY(lock_);
  236. };
  237. } // namespace base
  238. #endif // BASE_OBSERVER_LIST_THREADSAFE_H_