OLD | NEW |
---|---|
1 // Copyright 2016 The Chromium Authors. All rights reserved. | 1 // Copyright 2016 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 MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_ | 5 #ifndef MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_ |
6 #define MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_ | 6 #define MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_ |
7 | 7 |
8 #include <memory> | 8 #include <memory> |
9 | 9 |
10 #include "base/macros.h" | 10 #include "base/macros.h" |
11 #include "base/memory/ptr_util.h" | 11 #include "base/memory/ptr_util.h" |
12 #include "base/memory/ref_counted.h" | 12 #include "base/memory/ref_counted.h" |
13 #include "base/stl_util.h" | |
14 #include "base/synchronization/waitable_event.h" | |
13 #include "base/task_runner.h" | 15 #include "base/task_runner.h" |
14 #include "base/threading/thread_task_runner_handle.h" | 16 #include "base/threading/thread_task_runner_handle.h" |
15 #include "mojo/public/cpp/bindings/associated_group.h" | 17 #include "mojo/public/cpp/bindings/associated_group.h" |
16 #include "mojo/public/cpp/bindings/associated_interface_ptr.h" | 18 #include "mojo/public/cpp/bindings/associated_interface_ptr.h" |
17 #include "mojo/public/cpp/bindings/interface_ptr.h" | 19 #include "mojo/public/cpp/bindings/interface_ptr.h" |
18 #include "mojo/public/cpp/bindings/message.h" | 20 #include "mojo/public/cpp/bindings/message.h" |
21 #include "mojo/public/cpp/bindings/sync_call_restrictions.h" | |
22 #include "mojo/public/cpp/bindings/sync_event_watcher.h" | |
19 | 23 |
20 namespace mojo { | 24 namespace mojo { |
21 | 25 |
22 // Instances of this class may be used from any thread to serialize |Interface| | 26 // Instances of this class may be used from any thread to serialize |Interface| |
23 // messages and forward them elsewhere. In general you should use one of the | 27 // messages and forward them elsewhere. In general you should use one of the |
24 // ThreadSafeInterfacePtrBase helper aliases defined below, but this type may be | 28 // ThreadSafeInterfacePtrBase helper aliases defined below, but this type may be |
25 // useful if you need/want to manually manage the lifetime of the underlying | 29 // useful if you need/want to manually manage the lifetime of the underlying |
26 // proxy object which will be used to ultimately send messages. | 30 // proxy object which will be used to ultimately send messages. |
27 template <typename Interface> | 31 template <typename Interface> |
28 class ThreadSafeForwarder : public MessageReceiverWithResponder { | 32 class ThreadSafeForwarder : public MessageReceiverWithResponder { |
(...skipping 10 matching lines...) Expand all Loading... | |
39 // if any, back to the thread which called the corresponding interface method. | 43 // if any, back to the thread which called the corresponding interface method. |
40 ThreadSafeForwarder( | 44 ThreadSafeForwarder( |
41 const scoped_refptr<base::SingleThreadTaskRunner>& task_runner, | 45 const scoped_refptr<base::SingleThreadTaskRunner>& task_runner, |
42 const ForwardMessageCallback& forward, | 46 const ForwardMessageCallback& forward, |
43 const ForwardMessageWithResponderCallback& forward_with_responder, | 47 const ForwardMessageWithResponderCallback& forward_with_responder, |
44 const AssociatedGroup& associated_group) | 48 const AssociatedGroup& associated_group) |
45 : proxy_(this), | 49 : proxy_(this), |
46 task_runner_(task_runner), | 50 task_runner_(task_runner), |
47 forward_(forward), | 51 forward_(forward), |
48 forward_with_responder_(forward_with_responder), | 52 forward_with_responder_(forward_with_responder), |
49 associated_group_(associated_group) {} | 53 associated_group_(associated_group), |
54 sync_calls_(new InProgressSyncCalls()) {} | |
50 | 55 |
51 ~ThreadSafeForwarder() override {} | 56 ~ThreadSafeForwarder() override { |
57 // If there are ongoing sync calls signal their completion now. | |
58 base::AutoLock l(sync_calls_->lock); | |
59 for (const auto& pending_response : sync_calls_->pending_responses) | |
60 pending_response->event.Signal(); | |
61 } | |
52 | 62 |
53 ProxyType& proxy() { return proxy_; } | 63 ProxyType& proxy() { return proxy_; } |
54 | 64 |
55 private: | 65 private: |
56 // MessageReceiverWithResponder implementation: | 66 // MessageReceiverWithResponder implementation: |
57 bool Accept(Message* message) override { | 67 bool Accept(Message* message) override { |
58 if (!message->associated_endpoint_handles()->empty()) { | 68 if (!message->associated_endpoint_handles()->empty()) { |
59 // If this DCHECK fails, it is likely because: | 69 // If this DCHECK fails, it is likely because: |
60 // - This is a non-associated interface pointer setup using | 70 // - This is a non-associated interface pointer setup using |
61 // PtrWrapper::BindOnTaskRunner( | 71 // PtrWrapper::BindOnTaskRunner( |
62 // InterfacePtrInfo<InterfaceType> ptr_info); | 72 // InterfacePtrInfo<InterfaceType> ptr_info); |
63 // Please see the TODO in that method. | 73 // Please see the TODO in that method. |
64 // - This is an associated interface which hasn't been associated with a | 74 // - This is an associated interface which hasn't been associated with a |
65 // message pipe. In other words, the corresponding | 75 // message pipe. In other words, the corresponding |
66 // AssociatedInterfaceRequest hasn't been sent. | 76 // AssociatedInterfaceRequest hasn't been sent. |
67 DCHECK(associated_group_.GetController()); | 77 DCHECK(associated_group_.GetController()); |
68 message->SerializeAssociatedEndpointHandles( | 78 message->SerializeAssociatedEndpointHandles( |
69 associated_group_.GetController()); | 79 associated_group_.GetController()); |
70 } | 80 } |
71 task_runner_->PostTask(FROM_HERE, | 81 task_runner_->PostTask(FROM_HERE, |
72 base::Bind(forward_, base::Passed(message))); | 82 base::Bind(forward_, base::Passed(message))); |
73 return true; | 83 return true; |
74 } | 84 } |
75 | 85 |
76 bool AcceptWithResponder( | 86 bool AcceptWithResponder( |
77 Message* message, | 87 Message* message, |
78 std::unique_ptr<MessageReceiver> response_receiver) override { | 88 std::unique_ptr<MessageReceiver> responder) override { |
79 if (!message->associated_endpoint_handles()->empty()) { | 89 if (!message->associated_endpoint_handles()->empty()) { |
80 // Please see comment for the DCHECK in the previous method. | 90 // Please see comment for the DCHECK in the previous method. |
81 DCHECK(associated_group_.GetController()); | 91 DCHECK(associated_group_.GetController()); |
82 message->SerializeAssociatedEndpointHandles( | 92 message->SerializeAssociatedEndpointHandles( |
83 associated_group_.GetController()); | 93 associated_group_.GetController()); |
84 } | 94 } |
85 auto responder = | 95 |
86 base::MakeUnique<ForwardToCallingThread>(std::move(response_receiver)); | 96 // Async messages are always posted (even if |task_runner_| runs tasks on |
97 // this thread) to guarantee that two async calls can't be reordered. | |
98 if (!message->has_flag(Message::kFlagIsSync)) { | |
99 auto reply_forwarder = | |
100 base::MakeUnique<ForwardToCallingThread>(std::move(responder)); | |
101 task_runner_->PostTask( | |
102 FROM_HERE, base::Bind(forward_with_responder_, base::Passed(message), | |
103 base::Passed(&reply_forwarder))); | |
104 return true; | |
105 } | |
106 | |
107 SyncCallRestrictions::AssertSyncCallAllowed(); | |
108 | |
109 // If the InterfacePtr is bound to this thread, dispatch it directly. | |
110 if (task_runner_->RunsTasksOnCurrentThread()) { | |
111 forward_with_responder_.Run(std::move(*message), std::move(responder)); | |
112 return true; | |
113 } | |
114 | |
115 // If the InterfacePtr is bound on another thread, post the call. | |
116 // TODO(yzshen, watk): We block both this thread and the InterfacePtr | |
117 // thread. Ideally only this thread would block. | |
118 auto response = make_scoped_refptr(new SyncResponseInfo()); | |
119 auto response_signaler = base::MakeUnique<SyncResponseSignaler>(response); | |
87 task_runner_->PostTask( | 120 task_runner_->PostTask( |
88 FROM_HERE, base::Bind(forward_with_responder_, base::Passed(message), | 121 FROM_HERE, base::Bind(forward_with_responder_, base::Passed(message), |
89 base::Passed(&responder))); | 122 base::Passed(&response_signaler))); |
123 | |
124 // Save the pending SyncResponseInfo so that if the sync call deletes | |
125 // |this|, we can signal the completion of the call to return from | |
126 // SyncWatch(). | |
127 auto sync_calls = sync_calls_; | |
128 { | |
129 base::AutoLock l(sync_calls->lock); | |
130 sync_calls->pending_responses.push_back(response.get()); | |
131 } | |
132 | |
133 auto assign_true = [](bool* b) { *b = true; }; | |
134 bool event_signaled = false; | |
135 SyncEventWatcher watcher(&response->event, | |
136 base::Bind(assign_true, &event_signaled)); | |
137 watcher.SyncWatch(&event_signaled); | |
138 | |
139 { | |
140 base::AutoLock l(sync_calls->lock); | |
141 base::Erase(sync_calls->pending_responses, response.get()); | |
142 } | |
143 | |
144 if (event_signaled && response->received) | |
yzshen1
2017/03/30 20:53:06
Do we need to check |event_signaled|?
watk
2017/03/31 00:27:27
Can't think of a reason. Removed.
| |
145 ignore_result(responder->Accept(&response->message)); | |
146 | |
90 return true; | 147 return true; |
91 } | 148 } |
92 | 149 |
150 // Data that we need to share between the threads involved in a sync call. | |
151 struct SyncResponseInfo | |
152 : public base::RefCountedThreadSafe<SyncResponseInfo> { | |
153 Message message; | |
154 bool received = false; | |
155 base::WaitableEvent event{base::WaitableEvent::ResetPolicy::MANUAL, | |
156 base::WaitableEvent::InitialState::NOT_SIGNALED}; | |
157 | |
158 private: | |
159 friend class base::RefCountedThreadSafe<SyncResponseInfo>; | |
160 }; | |
161 | |
162 // A MessageReceiver that signals |response| when it either accepts the | |
163 // response message, or is destructed. | |
164 class SyncResponseSignaler : public MessageReceiver { | |
165 public: | |
166 explicit SyncResponseSignaler(scoped_refptr<SyncResponseInfo> response) | |
167 : response_(response) {} | |
168 | |
169 ~SyncResponseSignaler() override { | |
170 // If Accept() was not called we must still notify the waiter that the | |
171 // sync call is finished. | |
172 if (response_) | |
173 response_->event.Signal(); | |
174 } | |
175 | |
176 bool Accept(Message* message) { | |
177 response_->message = std::move(*message); | |
178 response_->received = true; | |
179 response_->event.Signal(); | |
180 response_ = nullptr; | |
181 return true; | |
182 } | |
183 | |
184 private: | |
185 scoped_refptr<SyncResponseInfo> response_; | |
186 }; | |
187 | |
188 // A record of the pending sync responses for canceling pending sync calls | |
189 // when the owning ThreadSafeForwarder is destructed. | |
190 struct InProgressSyncCalls | |
191 : public base::RefCountedThreadSafe<InProgressSyncCalls> { | |
192 // |lock| protects access to |pending_responses|. | |
193 base::Lock lock; | |
194 std::vector<SyncResponseInfo*> pending_responses; | |
195 }; | |
196 | |
93 class ForwardToCallingThread : public MessageReceiver { | 197 class ForwardToCallingThread : public MessageReceiver { |
94 public: | 198 public: |
95 explicit ForwardToCallingThread(std::unique_ptr<MessageReceiver> responder) | 199 explicit ForwardToCallingThread(std::unique_ptr<MessageReceiver> responder) |
96 : responder_(std::move(responder)), | 200 : responder_(std::move(responder)), |
97 caller_task_runner_(base::ThreadTaskRunnerHandle::Get()) { | 201 caller_task_runner_(base::ThreadTaskRunnerHandle::Get()) {} |
98 } | |
99 | 202 |
100 private: | 203 private: |
101 bool Accept(Message* message) { | 204 bool Accept(Message* message) { |
102 // The current instance will be deleted when this method returns, so we | 205 // The current instance will be deleted when this method returns, so we |
103 // have to relinquish the responder's ownership so it does not get | 206 // have to relinquish the responder's ownership so it does not get |
104 // deleted. | 207 // deleted. |
105 caller_task_runner_->PostTask(FROM_HERE, | 208 caller_task_runner_->PostTask( |
209 FROM_HERE, | |
106 base::Bind(&ForwardToCallingThread::CallAcceptAndDeleteResponder, | 210 base::Bind(&ForwardToCallingThread::CallAcceptAndDeleteResponder, |
107 base::Passed(std::move(responder_)), | 211 base::Passed(std::move(responder_)), |
108 base::Passed(std::move(*message)))); | 212 base::Passed(std::move(*message)))); |
109 return true; | 213 return true; |
110 } | 214 } |
111 | 215 |
112 static void CallAcceptAndDeleteResponder( | 216 static void CallAcceptAndDeleteResponder( |
113 std::unique_ptr<MessageReceiver> responder, | 217 std::unique_ptr<MessageReceiver> responder, |
114 Message message) { | 218 Message message) { |
115 ignore_result(responder->Accept(&message)); | 219 ignore_result(responder->Accept(&message)); |
116 } | 220 } |
117 | 221 |
118 std::unique_ptr<MessageReceiver> responder_; | 222 std::unique_ptr<MessageReceiver> responder_; |
119 scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner_; | 223 scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner_; |
120 }; | 224 }; |
121 | 225 |
122 ProxyType proxy_; | 226 ProxyType proxy_; |
123 const scoped_refptr<base::SingleThreadTaskRunner> task_runner_; | 227 const scoped_refptr<base::SingleThreadTaskRunner> task_runner_; |
124 const ForwardMessageCallback forward_; | 228 const ForwardMessageCallback forward_; |
125 const ForwardMessageWithResponderCallback forward_with_responder_; | 229 const ForwardMessageWithResponderCallback forward_with_responder_; |
126 AssociatedGroup associated_group_; | 230 AssociatedGroup associated_group_; |
231 scoped_refptr<InProgressSyncCalls> sync_calls_; | |
127 | 232 |
128 DISALLOW_COPY_AND_ASSIGN(ThreadSafeForwarder); | 233 DISALLOW_COPY_AND_ASSIGN(ThreadSafeForwarder); |
129 }; | 234 }; |
130 | 235 |
131 template <typename InterfacePtrType> | 236 template <typename InterfacePtrType> |
132 class ThreadSafeInterfacePtrBase | 237 class ThreadSafeInterfacePtrBase |
133 : public base::RefCountedThreadSafe< | 238 : public base::RefCountedThreadSafe< |
134 ThreadSafeInterfacePtrBase<InterfacePtrType>> { | 239 ThreadSafeInterfacePtrBase<InterfacePtrType>> { |
135 public: | 240 public: |
136 using InterfaceType = typename InterfacePtrType::InterfaceType; | 241 using InterfaceType = typename InterfacePtrType::InterfaceType; |
(...skipping 133 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
270 using ThreadSafeAssociatedInterfacePtr = | 375 using ThreadSafeAssociatedInterfacePtr = |
271 ThreadSafeInterfacePtrBase<AssociatedInterfacePtr<Interface>>; | 376 ThreadSafeInterfacePtrBase<AssociatedInterfacePtr<Interface>>; |
272 | 377 |
273 template <typename Interface> | 378 template <typename Interface> |
274 using ThreadSafeInterfacePtr = | 379 using ThreadSafeInterfacePtr = |
275 ThreadSafeInterfacePtrBase<InterfacePtr<Interface>>; | 380 ThreadSafeInterfacePtrBase<InterfacePtr<Interface>>; |
276 | 381 |
277 } // namespace mojo | 382 } // namespace mojo |
278 | 383 |
279 #endif // MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_ | 384 #endif // MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_ |
OLD | NEW |