OLD | NEW |
1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. | 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 | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #ifndef BASE_OBSERVER_LIST_THREADSAFE_H_ | 5 #ifndef BASE_OBSERVER_LIST_THREADSAFE_H_ |
6 #define BASE_OBSERVER_LIST_THREADSAFE_H_ | 6 #define BASE_OBSERVER_LIST_THREADSAFE_H_ |
7 | 7 |
8 #include <algorithm> | 8 #include <unordered_map> |
9 #include <map> | |
10 #include <memory> | |
11 #include <tuple> | |
12 | 9 |
13 #include "base/bind.h" | 10 #include "base/bind.h" |
14 #include "base/location.h" | 11 #include "base/location.h" |
15 #include "base/logging.h" | 12 #include "base/logging.h" |
16 #include "base/macros.h" | 13 #include "base/macros.h" |
17 #include "base/memory/ptr_util.h" | |
18 #include "base/memory/ref_counted.h" | 14 #include "base/memory/ref_counted.h" |
19 #include "base/observer_list.h" | 15 #include "base/observer_list.h" |
| 16 #include "base/sequenced_task_runner.h" |
| 17 #include "base/stl_util.h" |
| 18 #include "base/synchronization/lock.h" |
| 19 #include "base/threading/sequenced_task_runner_handle.h" |
| 20 #include "base/threading/thread_local.h" |
| 21 #include "build/build_config.h" |
| 22 |
| 23 // TODO(fdoray): Removing these includes causes IWYU failures in other headers, |
| 24 // remove them in a follow- up CL. |
| 25 #include "base/memory/ptr_util.h" |
20 #include "base/single_thread_task_runner.h" | 26 #include "base/single_thread_task_runner.h" |
21 #include "base/threading/platform_thread.h" | |
22 #include "base/threading/thread_task_runner_handle.h" | |
23 | 27 |
24 /////////////////////////////////////////////////////////////////////////////// | 28 /////////////////////////////////////////////////////////////////////////////// |
25 // | 29 // |
26 // OVERVIEW: | 30 // OVERVIEW: |
27 // | 31 // |
28 // A thread-safe container for a list of observers. | 32 // A thread-safe container for a list of observers. This is similar to the |
29 // This is similar to the observer_list (see observer_list.h), but it | 33 // observer_list (see observer_list.h), but it is more robust for multi- |
30 // is more robust for multi-threaded situations. | 34 // threaded situations. |
31 // | 35 // |
32 // The following use cases are supported: | 36 // The following use cases are supported: |
33 // * Observers can register for notifications from any thread. | 37 // * Observers can register for notifications from any sequence. They are |
34 // Callbacks to the observer will occur on the same thread where | 38 // always notified on the sequence from which they were registered. |
35 // the observer initially called AddObserver() from. | 39 // * Any sequence may trigger a notification via Notify(). |
36 // * Any thread may trigger a notification via Notify(). | 40 // * Observers can remove themselves from the observer list inside of a |
37 // * Observers can remove themselves from the observer list inside | 41 // callback. |
38 // of a callback. | 42 // * If one sequence is notifying observers concurrently with an observer |
39 // * If one thread is notifying observers concurrently with an observer | 43 // removing itself from the observer list, the notifications will be |
40 // removing itself from the observer list, the notifications will | 44 // silently dropped. |
41 // be silently dropped. | |
42 // | 45 // |
43 // The drawback of the threadsafe observer list is that notifications | 46 // The drawback of the threadsafe observer list is that notifications are not |
44 // are not as real-time as the non-threadsafe version of this class. | 47 // as real-time as the non-threadsafe version of this class. Notifications |
45 // Notifications will always be done via PostTask() to another thread, | 48 // will always be done via PostTask() to another sequence, whereas with the |
46 // whereas with the non-thread-safe observer_list, notifications happen | 49 // non-thread-safe observer_list, notifications happen synchronously. |
47 // synchronously and immediately. | |
48 // | |
49 // IMPLEMENTATION NOTES | |
50 // The ObserverListThreadSafe maintains an ObserverList for each thread | |
51 // which uses the ThreadSafeObserver. When Notifying the observers, | |
52 // we simply call PostTask to each registered thread, and then each thread | |
53 // will notify its regular ObserverList. | |
54 // | 50 // |
55 /////////////////////////////////////////////////////////////////////////////// | 51 /////////////////////////////////////////////////////////////////////////////// |
56 | 52 |
57 namespace base { | 53 namespace base { |
58 namespace internal { | 54 namespace internal { |
59 | 55 |
60 template <typename ObserverType, typename Method> | 56 template <typename ObserverType, typename Method> |
61 struct Dispatcher; | 57 struct Dispatcher; |
62 | 58 |
63 template <typename ObserverType, typename ReceiverType, typename... Params> | 59 template <typename ObserverType, typename ReceiverType, typename... Params> |
64 struct Dispatcher<ObserverType, void(ReceiverType::*)(Params...)> { | 60 struct Dispatcher<ObserverType, void(ReceiverType::*)(Params...)> { |
65 static void Run(void(ReceiverType::* m)(Params...), | 61 static void Run(void(ReceiverType::* m)(Params...), |
66 Params... params, ObserverType* obj) { | 62 Params... params, ObserverType* obj) { |
67 (obj->*m)(std::forward<Params>(params)...); | 63 (obj->*m)(std::forward<Params>(params)...); |
68 } | 64 } |
69 }; | 65 }; |
70 | 66 |
71 } // namespace internal | 67 } // namespace internal |
72 | 68 |
73 template <class ObserverType> | 69 template <class ObserverType> |
74 class ObserverListThreadSafe | 70 class ObserverListThreadSafe |
75 : public RefCountedThreadSafe<ObserverListThreadSafe<ObserverType>> { | 71 : public RefCountedThreadSafe<ObserverListThreadSafe<ObserverType>> { |
76 public: | 72 public: |
77 using NotificationType = | 73 using NotificationType = |
78 typename ObserverList<ObserverType>::NotificationType; | 74 typename ObserverList<ObserverType>::NotificationType; |
79 | 75 |
80 ObserverListThreadSafe() | 76 ObserverListThreadSafe() = default; |
81 : type_(ObserverListBase<ObserverType>::NOTIFY_ALL) {} | |
82 explicit ObserverListThreadSafe(NotificationType type) : type_(type) {} | 77 explicit ObserverListThreadSafe(NotificationType type) : type_(type) {} |
83 | 78 |
84 // Add an observer to the list. An observer should not be added to | 79 // Adds |observer| to the list. |observer| must not already be in the list. |
85 // the same list more than once. | 80 void AddObserver(ObserverType* observer) { |
86 void AddObserver(ObserverType* obs) { | 81 // TODO(fdoray): Change this to a DCHECK once all call sites have a |
87 // If there is no ThreadTaskRunnerHandle, it is impossible to notify on it, | 82 // SequencedTaskRunnerHandle. |
88 // so do not add the observer. | 83 if (!SequencedTaskRunnerHandle::IsSet()) |
89 if (!ThreadTaskRunnerHandle::IsSet()) | |
90 return; | 84 return; |
91 | 85 |
92 ObserverList<ObserverType>* list = nullptr; | 86 AutoLock auto_lock(lock_); |
93 PlatformThreadId thread_id = PlatformThread::CurrentId(); | 87 |
94 { | 88 // Add |observer| to the list of observers. |
95 AutoLock lock(list_lock_); | 89 DCHECK(!ContainsKey(observers_, observer)); |
96 if (observer_lists_.find(thread_id) == observer_lists_.end()) { | 90 const scoped_refptr<SequencedTaskRunner> task_runner = |
97 observer_lists_[thread_id] = | 91 SequencedTaskRunnerHandle::Get(); |
98 base::MakeUnique<ObserverListContext>(type_); | 92 observers_[observer] = task_runner; |
| 93 |
| 94 // If this is called while a notification is being dispatched on this thread |
| 95 // and |type_| is NOTIFY_ALL, |observer| must be notified (if a notification |
| 96 // is being dispatched on another thread in parallel, the notification may |
| 97 // or may not make it to |observer| depending on the outcome of the race to |
| 98 // |lock_|). |
| 99 if (type_ == NotificationType::NOTIFY_ALL) { |
| 100 const NotificationData* current_notification = |
| 101 tls_current_notification_.Get(); |
| 102 if (current_notification) { |
| 103 task_runner->PostTask( |
| 104 current_notification->from_here, |
| 105 Bind(&ObserverListThreadSafe<ObserverType>::NotifyWrapper, this, |
| 106 observer, *current_notification)); |
99 } | 107 } |
100 list = &(observer_lists_[thread_id]->list); | |
101 } | 108 } |
102 list->AddObserver(obs); | |
103 } | 109 } |
104 | 110 |
105 // Remove an observer from the list if it is in the list. | 111 // Remove an observer from the list if it is in the list. |
106 // If there are pending notifications in-transit to the observer, they will | 112 // If there are pending notifications in-transit to the observer, they will |
107 // be aborted. | 113 // be aborted. |
108 // If the observer to be removed is in the list, RemoveObserver MUST | 114 // If the observer to be removed is in the list, RemoveObserver MUST |
109 // be called from the same thread which called AddObserver. | 115 // be called from the same sequence which called AddObserver. |
110 void RemoveObserver(ObserverType* obs) { | 116 void RemoveObserver(ObserverType* observer) { |
111 PlatformThreadId thread_id = PlatformThread::CurrentId(); | 117 AutoLock auto_lock(lock_); |
112 { | 118 auto it = observers_.find(observer); |
113 AutoLock lock(list_lock_); | 119 if (it == observers_.end()) |
114 auto it = observer_lists_.find(thread_id); | 120 return; |
115 if (it == observer_lists_.end()) { | |
116 // This will happen if we try to remove an observer on a thread | |
117 // we never added an observer for. | |
118 return; | |
119 } | |
120 ObserverList<ObserverType>& list = it->second->list; | |
121 | 121 |
122 list.RemoveObserver(obs); | 122 // TODO(fdoray): Enable this on Android once all tests pass. |
| 123 #if !defined(OS_ANDROID) |
| 124 DCHECK(it->second->RunsTasksOnCurrentThread()); |
| 125 #endif |
123 | 126 |
124 // If that was the last observer in the list, remove the ObserverList | 127 observers_.erase(it); |
125 // entirely. | |
126 if (list.size() == 0) | |
127 observer_lists_.erase(it); | |
128 } | |
129 } | 128 } |
130 | 129 |
131 // Verifies that the list is currently empty (i.e. there are no observers). | 130 // Verifies that the list is currently empty (i.e. there are no observers). |
132 void AssertEmpty() const { | 131 void AssertEmpty() const { |
133 AutoLock lock(list_lock_); | 132 #if DCHECK_IS_ON() |
134 DCHECK(observer_lists_.empty()); | 133 AutoLock auto_lock(lock_); |
| 134 DCHECK(observers_.empty()); |
| 135 #endif |
135 } | 136 } |
136 | 137 |
137 // Notify methods. | 138 // Asynchronously invokes a callback on all observers, on their registration |
138 // Make a thread-safe callback to each Observer in the list. | 139 // sequence. You cannot assume that at the completion of the Notify call that |
139 // Note, these calls are effectively asynchronous. You cannot assume | 140 // all Observers have been Notified. The notification may still be pending |
140 // that at the completion of the Notify call that all Observers have | 141 // delivery. |
141 // been Notified. The notification may still be pending delivery. | |
142 template <typename Method, typename... Params> | 142 template <typename Method, typename... Params> |
143 void Notify(const tracked_objects::Location& from_here, | 143 void Notify(const tracked_objects::Location& from_here, |
144 Method m, Params&&... params) { | 144 Method m, Params&&... params) { |
145 Callback<void(ObserverType*)> method = | 145 Callback<void(ObserverType*)> method = |
146 Bind(&internal::Dispatcher<ObserverType, Method>::Run, | 146 Bind(&internal::Dispatcher<ObserverType, Method>::Run, |
147 m, std::forward<Params>(params)...); | 147 m, std::forward<Params>(params)...); |
148 | 148 |
149 AutoLock lock(list_lock_); | 149 AutoLock lock(lock_); |
150 for (const auto& entry : observer_lists_) { | 150 for (const auto& observer : observers_) { |
151 ObserverListContext* context = entry.second.get(); | 151 observer.second->PostTask( |
152 context->task_runner->PostTask( | |
153 from_here, | 152 from_here, |
154 Bind(&ObserverListThreadSafe<ObserverType>::NotifyWrapper, | 153 Bind(&ObserverListThreadSafe<ObserverType>::NotifyWrapper, this, |
155 this, context, method)); | 154 observer.first, NotificationData(from_here, method))); |
156 } | 155 } |
157 } | 156 } |
158 | 157 |
159 private: | 158 private: |
160 friend class RefCountedThreadSafe<ObserverListThreadSafe<ObserverType>>; | 159 friend class RefCountedThreadSafe<ObserverListThreadSafe<ObserverType>>; |
161 | 160 |
162 struct ObserverListContext { | 161 struct NotificationData { |
163 explicit ObserverListContext(NotificationType type) | 162 NotificationData(const tracked_objects::Location& from_here_in, |
164 : task_runner(ThreadTaskRunnerHandle::Get()), list(type) {} | 163 const Callback<void(ObserverType*)>& method_in) |
| 164 : from_here(from_here_in), method(method_in) {} |
165 | 165 |
166 scoped_refptr<SingleThreadTaskRunner> task_runner; | 166 tracked_objects::Location from_here; |
167 ObserverList<ObserverType> list; | 167 Callback<void(ObserverType*)> method; |
168 | |
169 private: | |
170 DISALLOW_COPY_AND_ASSIGN(ObserverListContext); | |
171 }; | 168 }; |
172 | 169 |
173 ~ObserverListThreadSafe() { | 170 ~ObserverListThreadSafe() = default; |
| 171 |
| 172 void NotifyWrapper(ObserverType* observer, |
| 173 const NotificationData& notification) { |
| 174 { |
| 175 AutoLock auto_lock(lock_); |
| 176 |
| 177 // Check whether the observer still needs a notification. |
| 178 auto it = observers_.find(observer); |
| 179 if (it == observers_.end()) |
| 180 return; |
| 181 DCHECK(it->second->RunsTasksOnCurrentThread()); |
| 182 } |
| 183 |
| 184 // Keep track of the notification being dispatched on the current thread. |
| 185 // This will be used if the callback below calls AddObserver(). |
| 186 // |
| 187 // Note: |tls_current_notification_| may not be nullptr if this runs in a |
| 188 // nested loop started by a notification callback. In that case, it is |
| 189 // important to save the previous value to restore it later. |
| 190 const NotificationData* const previous_notification = |
| 191 tls_current_notification_.Get(); |
| 192 tls_current_notification_.Set(¬ification); |
| 193 |
| 194 // Invoke the callback. |
| 195 notification.method.Run(observer); |
| 196 |
| 197 // Reset the notification being dispatched on the current thread to its |
| 198 // previous value. |
| 199 tls_current_notification_.Set(previous_notification); |
174 } | 200 } |
175 | 201 |
176 // Wrapper which is called to fire the notifications for each thread's | 202 const NotificationType type_ = NotificationType::NOTIFY_ALL; |
177 // ObserverList. This function MUST be called on the thread which owns | |
178 // the unsafe ObserverList. | |
179 void NotifyWrapper(ObserverListContext* context, | |
180 const Callback<void(ObserverType*)>& method) { | |
181 // Check that this list still needs notifications. | |
182 { | |
183 AutoLock lock(list_lock_); | |
184 auto it = observer_lists_.find(PlatformThread::CurrentId()); | |
185 | 203 |
186 // The ObserverList could have been removed already. In fact, it could | 204 // Synchronizes access to |observers_|. |
187 // have been removed and then re-added! If the master list's loop | 205 mutable Lock lock_; |
188 // does not match this one, then we do not need to finish this | |
189 // notification. | |
190 if (it == observer_lists_.end() || it->second.get() != context) | |
191 return; | |
192 } | |
193 | 206 |
194 for (auto& observer : context->list) { | 207 // Keys are observers. Values are the SequencedTaskRunners on which they must |
195 method.Run(&observer); | 208 // be notified. |
196 } | 209 std::unordered_map<ObserverType*, scoped_refptr<SequencedTaskRunner>> |
| 210 observers_; |
197 | 211 |
198 // If there are no more observers on the list, we can now delete it. | 212 // Notification being dispatched on the current thread. |
199 if (context->list.size() == 0) { | 213 ThreadLocalPointer<const NotificationData> tls_current_notification_; |
200 { | |
201 AutoLock lock(list_lock_); | |
202 // Remove |list| if it's not already removed. | |
203 // This can happen if multiple observers got removed in a notification. | |
204 // See http://crbug.com/55725. | |
205 auto it = observer_lists_.find(PlatformThread::CurrentId()); | |
206 if (it != observer_lists_.end() && it->second.get() == context) | |
207 observer_lists_.erase(it); | |
208 } | |
209 } | |
210 } | |
211 | |
212 mutable Lock list_lock_; // Protects the observer_lists_. | |
213 | |
214 // Key by PlatformThreadId because in tests, clients can attempt to remove | |
215 // observers without a SingleThreadTaskRunner. If this were keyed by | |
216 // SingleThreadTaskRunner, that operation would be silently ignored, leaving | |
217 // garbage in the ObserverList. | |
218 std::map<PlatformThreadId, std::unique_ptr<ObserverListContext>> | |
219 observer_lists_; | |
220 | |
221 const NotificationType type_; | |
222 | 214 |
223 DISALLOW_COPY_AND_ASSIGN(ObserverListThreadSafe); | 215 DISALLOW_COPY_AND_ASSIGN(ObserverListThreadSafe); |
224 }; | 216 }; |
225 | 217 |
226 } // namespace base | 218 } // namespace base |
227 | 219 |
228 #endif // BASE_OBSERVER_LIST_THREADSAFE_H_ | 220 #endif // BASE_OBSERVER_LIST_THREADSAFE_H_ |
OLD | NEW |