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
204
205
206
207
208
209
|
// Copyright (c) 2010 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.h"
#include "base/logging.h"
#include "base/message_loop.h"
#include "base/observer_list.h"
#include "base/ref_counted.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.
//
// 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 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:
typedef typename ObserverList<ObserverType>::NotificationType
NotificationType;
ObserverListThreadSafe()
: type_(ObserverListBase<ObserverType>::NOTIFY_ALL) {}
explicit ObserverListThreadSafe(NotificationType type) : type_(type) {}
~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>(type_);
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) {
{
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());
if (it != observer_lists_.end() && it->second == list)
observer_lists_.erase(it);
}
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_;
const NotificationType type_;
DISALLOW_COPY_AND_ASSIGN(ObserverListThreadSafe);
};
#endif // BASE_OBSERVER_LIST_THREADSAFE_H_
|
