OLD | NEW |
1 // Copyright (c) 2011 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2011 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 #include "ipc/ipc_sync_channel.h" | 5 #include "ipc/ipc_sync_channel.h" |
6 | 6 |
7 #include "base/lazy_instance.h" | 7 #include "base/lazy_instance.h" |
8 #include "base/logging.h" | 8 #include "base/logging.h" |
9 #include "base/message_loop.h" | |
10 #include "base/threading/thread_local.h" | 9 #include "base/threading/thread_local.h" |
11 #include "base/synchronization/waitable_event.h" | 10 #include "base/synchronization/waitable_event.h" |
12 #include "base/synchronization/waitable_event_watcher.h" | 11 #include "base/synchronization/waitable_event_watcher.h" |
13 #include "ipc/ipc_sync_message.h" | 12 #include "ipc/ipc_sync_message.h" |
14 | 13 |
15 using base::TimeDelta; | 14 using base::TimeDelta; |
16 using base::TimeTicks; | 15 using base::TimeTicks; |
17 using base::WaitableEvent; | 16 using base::WaitableEvent; |
18 | 17 |
19 namespace IPC { | 18 namespace IPC { |
(...skipping 107 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
127 } | 126 } |
128 } | 127 } |
129 | 128 |
130 if (--listener_count_ == 0) { | 129 if (--listener_count_ == 0) { |
131 DCHECK(lazy_tls_ptr_.Pointer()->Get()); | 130 DCHECK(lazy_tls_ptr_.Pointer()->Get()); |
132 lazy_tls_ptr_.Pointer()->Set(NULL); | 131 lazy_tls_ptr_.Pointer()->Set(NULL); |
133 } | 132 } |
134 } | 133 } |
135 | 134 |
136 WaitableEvent* dispatch_event() { return &dispatch_event_; } | 135 WaitableEvent* dispatch_event() { return &dispatch_event_; } |
137 MessageLoop* listener_message_loop() { return listener_message_loop_; } | 136 base::MessageLoopProxy* listener_message_loop() { |
| 137 return listener_message_loop_; |
| 138 } |
138 | 139 |
139 // Holds a pointer to the per-thread ReceivedSyncMsgQueue object. | 140 // Holds a pointer to the per-thread ReceivedSyncMsgQueue object. |
140 static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> > | 141 static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> > |
141 lazy_tls_ptr_; | 142 lazy_tls_ptr_; |
142 | 143 |
143 // Called on the ipc thread to check if we can unblock any current Send() | 144 // Called on the ipc thread to check if we can unblock any current Send() |
144 // calls based on a queued reply. | 145 // calls based on a queued reply. |
145 void DispatchReplies() { | 146 void DispatchReplies() { |
146 for (size_t i = 0; i < received_replies_.size(); ++i) { | 147 for (size_t i = 0; i < received_replies_.size(); ++i) { |
147 Message* message = received_replies_[i].message; | 148 Message* message = received_replies_[i].message; |
(...skipping 13 matching lines...) Expand all Loading... |
161 top_send_done_watcher_ = watcher; | 162 top_send_done_watcher_ = watcher; |
162 } | 163 } |
163 | 164 |
164 private: | 165 private: |
165 friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>; | 166 friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>; |
166 | 167 |
167 // See the comment in SyncChannel::SyncChannel for why this event is created | 168 // See the comment in SyncChannel::SyncChannel for why this event is created |
168 // as manual reset. | 169 // as manual reset. |
169 ReceivedSyncMsgQueue() : | 170 ReceivedSyncMsgQueue() : |
170 dispatch_event_(true, false), | 171 dispatch_event_(true, false), |
171 listener_message_loop_(MessageLoop::current()), | 172 listener_message_loop_(base::MessageLoopProxy::CreateForCurrentThread()), |
172 task_pending_(false), | 173 task_pending_(false), |
173 listener_count_(0), | 174 listener_count_(0), |
174 top_send_done_watcher_(NULL) { | 175 top_send_done_watcher_(NULL) { |
175 } | 176 } |
176 | 177 |
177 ~ReceivedSyncMsgQueue() {} | 178 ~ReceivedSyncMsgQueue() {} |
178 | 179 |
179 // Holds information about a queued synchronous message or reply. | 180 // Holds information about a queued synchronous message or reply. |
180 struct QueuedMessage { | 181 struct QueuedMessage { |
181 QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { } | 182 QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { } |
182 Message* message; | 183 Message* message; |
183 scoped_refptr<SyncChannel::SyncContext> context; | 184 scoped_refptr<SyncChannel::SyncContext> context; |
184 }; | 185 }; |
185 | 186 |
186 typedef std::deque<QueuedMessage> SyncMessageQueue; | 187 typedef std::deque<QueuedMessage> SyncMessageQueue; |
187 SyncMessageQueue message_queue_; | 188 SyncMessageQueue message_queue_; |
188 | 189 |
189 std::vector<QueuedMessage> received_replies_; | 190 std::vector<QueuedMessage> received_replies_; |
190 | 191 |
191 // Set when we got a synchronous message that we must respond to as the | 192 // Set when we got a synchronous message that we must respond to as the |
192 // sender needs its reply before it can reply to our original synchronous | 193 // sender needs its reply before it can reply to our original synchronous |
193 // message. | 194 // message. |
194 WaitableEvent dispatch_event_; | 195 WaitableEvent dispatch_event_; |
195 MessageLoop* listener_message_loop_; | 196 scoped_refptr<base::MessageLoopProxy> listener_message_loop_; |
196 base::Lock message_lock_; | 197 base::Lock message_lock_; |
197 bool task_pending_; | 198 bool task_pending_; |
198 int listener_count_; | 199 int listener_count_; |
199 | 200 |
200 // The current send done event watcher for this thread. Used to maintain | 201 // The current send done event watcher for this thread. Used to maintain |
201 // a local global stack of send done watchers to ensure that nested sync | 202 // a local global stack of send done watchers to ensure that nested sync |
202 // message loops complete correctly. | 203 // message loops complete correctly. |
203 base::WaitableEventWatcher* top_send_done_watcher_; | 204 base::WaitableEventWatcher* top_send_done_watcher_; |
204 }; | 205 }; |
205 | 206 |
206 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> > | 207 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> > |
207 SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_(base::LINKER_INITIALIZED); | 208 SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_(base::LINKER_INITIALIZED); |
208 | 209 |
209 SyncChannel::SyncContext::SyncContext( | 210 SyncChannel::SyncContext::SyncContext( |
210 Channel::Listener* listener, | 211 Channel::Listener* listener, |
211 MessageLoop* ipc_thread, | 212 base::MessageLoopProxy* ipc_thread, |
212 WaitableEvent* shutdown_event) | 213 WaitableEvent* shutdown_event) |
213 : ChannelProxy::Context(listener, ipc_thread), | 214 : ChannelProxy::Context(listener, ipc_thread), |
214 received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()), | 215 received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()), |
215 shutdown_event_(shutdown_event), | 216 shutdown_event_(shutdown_event), |
216 restrict_dispatch_(false) { | 217 restrict_dispatch_(false) { |
217 } | 218 } |
218 | 219 |
219 SyncChannel::SyncContext::~SyncContext() { | 220 SyncChannel::SyncContext::~SyncContext() { |
220 while (!deserializers_.empty()) | 221 while (!deserializers_.empty()) |
221 Pop(); | 222 Pop(); |
(...skipping 142 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
364 DCHECK_EQ(GetSendDoneEvent(), event); | 365 DCHECK_EQ(GetSendDoneEvent(), event); |
365 MessageLoop::current()->QuitNow(); | 366 MessageLoop::current()->QuitNow(); |
366 } | 367 } |
367 } | 368 } |
368 | 369 |
369 | 370 |
370 SyncChannel::SyncChannel( | 371 SyncChannel::SyncChannel( |
371 const IPC::ChannelHandle& channel_handle, | 372 const IPC::ChannelHandle& channel_handle, |
372 Channel::Mode mode, | 373 Channel::Mode mode, |
373 Channel::Listener* listener, | 374 Channel::Listener* listener, |
374 MessageLoop* ipc_message_loop, | 375 base::MessageLoopProxy* ipc_message_loop, |
375 bool create_pipe_now, | 376 bool create_pipe_now, |
376 WaitableEvent* shutdown_event) | 377 WaitableEvent* shutdown_event) |
377 : ChannelProxy( | 378 : ChannelProxy( |
378 channel_handle, mode, ipc_message_loop, | 379 channel_handle, mode, ipc_message_loop, |
379 new SyncContext(listener, ipc_message_loop, shutdown_event), | 380 new SyncContext(listener, ipc_message_loop, shutdown_event), |
380 create_pipe_now), | 381 create_pipe_now), |
381 sync_messages_with_no_timeout_allowed_(true) { | 382 sync_messages_with_no_timeout_allowed_(true) { |
382 // Ideally we only want to watch this object when running a nested message | 383 // Ideally we only want to watch this object when running a nested message |
383 // loop. However, we don't know when it exits if there's another nested | 384 // loop. However, we don't know when it exits if there's another nested |
384 // message loop running under it or not, so we wouldn't know whether to | 385 // message loop running under it or not, so we wouldn't know whether to |
(...skipping 118 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
503 void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) { | 504 void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) { |
504 DCHECK(event == sync_context()->GetDispatchEvent()); | 505 DCHECK(event == sync_context()->GetDispatchEvent()); |
505 // The call to DispatchMessages might delete this object, so reregister | 506 // The call to DispatchMessages might delete this object, so reregister |
506 // the object watcher first. | 507 // the object watcher first. |
507 event->Reset(); | 508 event->Reset(); |
508 dispatch_watcher_.StartWatching(event, this); | 509 dispatch_watcher_.StartWatching(event, this); |
509 sync_context()->DispatchMessages(); | 510 sync_context()->DispatchMessages(); |
510 } | 511 } |
511 | 512 |
512 } // namespace IPC | 513 } // namespace IPC |
OLD | NEW |