Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(53)

Side by Side Diff: chrome/common/ipc_sync_channel.cc

Issue 155905: Separates ipc code from common (http://crbug.com/16829) (Closed)
Patch Set: Fixes reference to 'common_message_traits' it's actually 'common_param_traits' Created 11 years, 5 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
« no previous file with comments | « chrome/common/ipc_sync_channel.h ('k') | chrome/common/ipc_sync_channel_unittest.cc » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
(Empty)
1 // Copyright (c) 2006-2008 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
5 #include "chrome/common/ipc_sync_channel.h"
6
7 #include "base/lazy_instance.h"
8 #include "base/logging.h"
9 #include "base/thread_local.h"
10 #include "base/message_loop.h"
11 #include "base/waitable_event.h"
12 #include "base/waitable_event_watcher.h"
13 #include "chrome/common/ipc_sync_message.h"
14
15 using base::TimeDelta;
16 using base::TimeTicks;
17 using base::WaitableEvent;
18
19 namespace IPC {
20 // When we're blocked in a Send(), we need to process incoming synchronous
21 // messages right away because it could be blocking our reply (either
22 // directly from the same object we're calling, or indirectly through one or
23 // more other channels). That means that in SyncContext's OnMessageReceived,
24 // we need to process sync message right away if we're blocked. However a
25 // simple check isn't sufficient, because the listener thread can be in the
26 // process of calling Send.
27 // To work around this, when SyncChannel filters a sync message, it sets
28 // an event that the listener thread waits on during its Send() call. This
29 // allows us to dispatch incoming sync messages when blocked. The race
30 // condition is handled because if Send is in the process of being called, it
31 // will check the event. In case the listener thread isn't sending a message,
32 // we queue a task on the listener thread to dispatch the received messages.
33 // The messages are stored in this queue object that's shared among all
34 // SyncChannel objects on the same thread (since one object can receive a
35 // sync message while another one is blocked).
36
37 class SyncChannel::ReceivedSyncMsgQueue :
38 public base::RefCountedThreadSafe<ReceivedSyncMsgQueue> {
39 public:
40 // Returns the ReceivedSyncMsgQueue instance for this thread, creating one
41 // if necessary. Call RemoveContext on the same thread when done.
42 static ReceivedSyncMsgQueue* AddContext() {
43 // We want one ReceivedSyncMsgQueue per listener thread (i.e. since multiple
44 // SyncChannel objects can block the same thread).
45 ReceivedSyncMsgQueue* rv = lazy_tls_ptr_.Pointer()->Get();
46 if (!rv) {
47 rv = new ReceivedSyncMsgQueue();
48 ReceivedSyncMsgQueue::lazy_tls_ptr_.Pointer()->Set(rv);
49 }
50 rv->listener_count_++;
51 return rv;
52 }
53
54 ~ReceivedSyncMsgQueue() {
55 }
56
57 // Called on IPC thread when a synchronous message or reply arrives.
58 void QueueMessage(const Message& msg, SyncChannel::SyncContext* context) {
59 bool was_task_pending;
60 {
61 AutoLock auto_lock(message_lock_);
62
63 was_task_pending = task_pending_;
64 task_pending_ = true;
65
66 // We set the event in case the listener thread is blocked (or is about
67 // to). In case it's not, the PostTask dispatches the messages.
68 message_queue_.push_back(QueuedMessage(new Message(msg), context));
69 }
70
71 dispatch_event_.Signal();
72 if (!was_task_pending) {
73 listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
74 this, &ReceivedSyncMsgQueue::DispatchMessagesTask));
75 }
76 }
77
78 void QueueReply(const Message &msg, SyncChannel::SyncContext* context) {
79 received_replies_.push_back(QueuedMessage(new Message(msg), context));
80 }
81
82 // Called on the listener's thread to process any queues synchronous
83 // messages.
84 void DispatchMessagesTask() {
85 {
86 AutoLock auto_lock(message_lock_);
87 task_pending_ = false;
88 }
89 DispatchMessages();
90 }
91
92 void DispatchMessages() {
93 while (true) {
94 Message* message;
95 scoped_refptr<SyncChannel::SyncContext> context;
96 {
97 AutoLock auto_lock(message_lock_);
98 if (message_queue_.empty())
99 break;
100
101 message = message_queue_.front().message;
102 context = message_queue_.front().context;
103 message_queue_.pop_front();
104 }
105
106 context->OnDispatchMessage(*message);
107 delete message;
108 }
109 }
110
111 // SyncChannel calls this in its destructor.
112 void RemoveContext(SyncContext* context) {
113 AutoLock auto_lock(message_lock_);
114
115 SyncMessageQueue::iterator iter = message_queue_.begin();
116 while (iter != message_queue_.end()) {
117 if (iter->context == context) {
118 delete iter->message;
119 iter = message_queue_.erase(iter);
120 } else {
121 iter++;
122 }
123 }
124
125 if (--listener_count_ == 0) {
126 DCHECK(lazy_tls_ptr_.Pointer()->Get());
127 lazy_tls_ptr_.Pointer()->Set(NULL);
128 }
129 }
130
131 WaitableEvent* dispatch_event() { return &dispatch_event_; }
132 MessageLoop* listener_message_loop() { return listener_message_loop_; }
133
134 // Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
135 static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> >
136 lazy_tls_ptr_;
137
138 // Called on the ipc thread to check if we can unblock any current Send()
139 // calls based on a queued reply.
140 void DispatchReplies() {
141 for (size_t i = 0; i < received_replies_.size(); ++i) {
142 Message* message = received_replies_[i].message;
143 if (received_replies_[i].context->TryToUnblockListener(message)) {
144 delete message;
145 received_replies_.erase(received_replies_.begin() + i);
146 return;
147 }
148 }
149 }
150
151 private:
152 // See the comment in SyncChannel::SyncChannel for why this event is created
153 // as manual reset.
154 ReceivedSyncMsgQueue() :
155 dispatch_event_(true, false),
156 listener_message_loop_(MessageLoop::current()),
157 task_pending_(false),
158 listener_count_(0) {
159 }
160
161 // Holds information about a queued synchronous message or reply.
162 struct QueuedMessage {
163 QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
164 Message* message;
165 scoped_refptr<SyncChannel::SyncContext> context;
166 };
167
168 typedef std::deque<QueuedMessage> SyncMessageQueue;
169 SyncMessageQueue message_queue_;
170
171 std::vector<QueuedMessage> received_replies_;
172
173 // Set when we got a synchronous message that we must respond to as the
174 // sender needs its reply before it can reply to our original synchronous
175 // message.
176 WaitableEvent dispatch_event_;
177 MessageLoop* listener_message_loop_;
178 Lock message_lock_;
179 bool task_pending_;
180 int listener_count_;
181 };
182
183 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> >
184 SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_(base::LINKER_INITIALIZED);
185
186 SyncChannel::SyncContext::SyncContext(
187 Channel::Listener* listener,
188 MessageFilter* filter,
189 MessageLoop* ipc_thread,
190 WaitableEvent* shutdown_event)
191 : ChannelProxy::Context(listener, filter, ipc_thread),
192 received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
193 shutdown_event_(shutdown_event) {
194 }
195
196 SyncChannel::SyncContext::~SyncContext() {
197 while (!deserializers_.empty())
198 Pop();
199 }
200
201 // Adds information about an outgoing sync message to the context so that
202 // we know how to deserialize the reply. Returns a handle that's set when
203 // the reply has arrived.
204 void SyncChannel::SyncContext::Push(SyncMessage* sync_msg) {
205 // The event is created as manual reset because in between Signal and
206 // OnObjectSignalled, another Send can happen which would stop the watcher
207 // from being called. The event would get watched later, when the nested
208 // Send completes, so the event will need to remain set.
209 PendingSyncMsg pending(SyncMessage::GetMessageId(*sync_msg),
210 sync_msg->GetReplyDeserializer(),
211 new WaitableEvent(true, false));
212 AutoLock auto_lock(deserializers_lock_);
213 deserializers_.push_back(pending);
214 }
215
216 bool SyncChannel::SyncContext::Pop() {
217 bool result;
218 {
219 AutoLock auto_lock(deserializers_lock_);
220 PendingSyncMsg msg = deserializers_.back();
221 delete msg.deserializer;
222 delete msg.done_event;
223 msg.done_event = NULL;
224 deserializers_.pop_back();
225 result = msg.send_result;
226 }
227
228 // We got a reply to a synchronous Send() call that's blocking the listener
229 // thread. However, further down the call stack there could be another
230 // blocking Send() call, whose reply we received after we made this last
231 // Send() call. So check if we have any queued replies available that
232 // can now unblock the listener thread.
233 ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
234 received_sync_msgs_.get(), &ReceivedSyncMsgQueue::DispatchReplies));
235
236 return result;
237 }
238
239 WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
240 AutoLock auto_lock(deserializers_lock_);
241 return deserializers_.back().done_event;
242 }
243
244 WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
245 return received_sync_msgs_->dispatch_event();
246 }
247
248 void SyncChannel::SyncContext::DispatchMessages() {
249 received_sync_msgs_->DispatchMessages();
250 }
251
252 bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
253 AutoLock auto_lock(deserializers_lock_);
254 if (deserializers_.empty() ||
255 !SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
256 return false;
257 }
258
259 if (!msg->is_reply_error()) {
260 deserializers_.back().send_result = deserializers_.back().deserializer->
261 SerializeOutputParameters(*msg);
262 }
263 deserializers_.back().done_event->Signal();
264
265 return true;
266 }
267
268 void SyncChannel::SyncContext::Clear() {
269 CancelPendingSends();
270 received_sync_msgs_->RemoveContext(this);
271
272 Context::Clear();
273 }
274
275 void SyncChannel::SyncContext::OnMessageReceived(const Message& msg) {
276 // Give the filters a chance at processing this message.
277 if (TryFilters(msg))
278 return;
279
280 if (TryToUnblockListener(&msg))
281 return;
282
283 if (msg.should_unblock()) {
284 received_sync_msgs_->QueueMessage(msg, this);
285 return;
286 }
287
288 if (msg.is_reply()) {
289 received_sync_msgs_->QueueReply(msg, this);
290 return;
291 }
292
293 return Context::OnMessageReceivedNoFilter(msg);
294 }
295
296 void SyncChannel::SyncContext::OnChannelError() {
297 CancelPendingSends();
298 shutdown_watcher_.StopWatching();
299 Context::OnChannelError();
300 }
301
302 void SyncChannel::SyncContext::OnChannelOpened() {
303 shutdown_watcher_.StartWatching(shutdown_event_, this);
304 Context::OnChannelOpened();
305 }
306
307 void SyncChannel::SyncContext::OnChannelClosed() {
308 shutdown_watcher_.StopWatching();
309 Context::OnChannelClosed();
310 }
311
312 void SyncChannel::SyncContext::OnSendTimeout(int message_id) {
313 AutoLock auto_lock(deserializers_lock_);
314 PendingSyncMessageQueue::iterator iter;
315 for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
316 if (iter->id == message_id) {
317 iter->done_event->Signal();
318 break;
319 }
320 }
321 }
322
323 void SyncChannel::SyncContext::CancelPendingSends() {
324 AutoLock auto_lock(deserializers_lock_);
325 PendingSyncMessageQueue::iterator iter;
326 for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++)
327 iter->done_event->Signal();
328 }
329
330 void SyncChannel::SyncContext::OnWaitableEventSignaled(WaitableEvent* event) {
331 DCHECK(event == shutdown_event_);
332 // Process shut down before we can get a reply to a synchronous message.
333 // Cancel pending Send calls, which will end up setting the send done event.
334 CancelPendingSends();
335 }
336
337
338 SyncChannel::SyncChannel(
339 const std::string& channel_id, Channel::Mode mode,
340 Channel::Listener* listener, MessageFilter* filter,
341 MessageLoop* ipc_message_loop, bool create_pipe_now,
342 WaitableEvent* shutdown_event)
343 : ChannelProxy(
344 channel_id, mode, ipc_message_loop,
345 new SyncContext(listener, filter, ipc_message_loop, shutdown_event),
346 create_pipe_now),
347 sync_messages_with_no_timeout_allowed_(true) {
348 // Ideally we only want to watch this object when running a nested message
349 // loop. However, we don't know when it exits if there's another nested
350 // message loop running under it or not, so we wouldn't know whether to
351 // stop or keep watching. So we always watch it, and create the event as
352 // manual reset since the object watcher might otherwise reset the event
353 // when we're doing a WaitMany.
354 dispatch_watcher_.StartWatching(sync_context()->GetDispatchEvent(), this);
355 }
356
357 SyncChannel::~SyncChannel() {
358 }
359
360 bool SyncChannel::Send(Message* message) {
361 return SendWithTimeout(message, base::kNoTimeout);
362 }
363
364 bool SyncChannel::SendWithTimeout(Message* message, int timeout_ms) {
365 if (!message->is_sync()) {
366 ChannelProxy::Send(message);
367 return true;
368 }
369
370 // *this* might get deleted in WaitForReply.
371 scoped_refptr<SyncContext> context(sync_context());
372 if (context->shutdown_event()->IsSignaled()) {
373 delete message;
374 return false;
375 }
376
377 DCHECK(sync_messages_with_no_timeout_allowed_ ||
378 timeout_ms != base::kNoTimeout);
379 SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
380 context->Push(sync_msg);
381 int message_id = SyncMessage::GetMessageId(*sync_msg);
382 WaitableEvent* pump_messages_event = sync_msg->pump_messages_event();
383
384 ChannelProxy::Send(message);
385
386 if (timeout_ms != base::kNoTimeout) {
387 // We use the sync message id so that when a message times out, we don't
388 // confuse it with another send that is either above/below this Send in
389 // the call stack.
390 context->ipc_message_loop()->PostDelayedTask(FROM_HERE,
391 NewRunnableMethod(context.get(),
392 &SyncContext::OnSendTimeout, message_id), timeout_ms);
393 }
394
395 // Wait for reply, or for any other incoming synchronous messages.
396 WaitForReply(pump_messages_event);
397
398 return context->Pop();
399 }
400
401 void SyncChannel::WaitForReply(WaitableEvent* pump_messages_event) {
402 while (true) {
403 WaitableEvent* objects[] = {
404 sync_context()->GetDispatchEvent(),
405 sync_context()->GetSendDoneEvent(),
406 pump_messages_event
407 };
408
409 unsigned count = pump_messages_event ? 3: 2;
410 unsigned result = WaitableEvent::WaitMany(objects, count);
411 if (result == 0 /* dispatch event */) {
412 // We're waiting for a reply, but we received a blocking synchronous
413 // call. We must process it or otherwise a deadlock might occur.
414 sync_context()->GetDispatchEvent()->Reset();
415 sync_context()->DispatchMessages();
416 continue;
417 }
418
419 if (result == 2 /* pump_messages_event */)
420 WaitForReplyWithNestedMessageLoop(); // Start a nested message loop.
421
422 break;
423 }
424 }
425
426 void SyncChannel::WaitForReplyWithNestedMessageLoop() {
427 WaitableEvent* old_done_event = send_done_watcher_.GetWatchedEvent();
428 send_done_watcher_.StopWatching();
429 send_done_watcher_.StartWatching(sync_context()->GetSendDoneEvent(), this);
430 bool old_state = MessageLoop::current()->NestableTasksAllowed();
431 MessageLoop::current()->SetNestableTasksAllowed(true);
432 MessageLoop::current()->Run();
433 MessageLoop::current()->SetNestableTasksAllowed(old_state);
434 if (old_done_event)
435 send_done_watcher_.StartWatching(old_done_event, this);
436 }
437
438 void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) {
439 WaitableEvent* dispatch_event = sync_context()->GetDispatchEvent();
440 if (event == dispatch_event) {
441 // The call to DispatchMessages might delete this object, so reregister
442 // the object watcher first.
443 dispatch_event->Reset();
444 dispatch_watcher_.StartWatching(dispatch_event, this);
445 sync_context()->DispatchMessages();
446 } else {
447 // We got the reply, timed out or the process shutdown.
448 DCHECK(event == sync_context()->GetSendDoneEvent());
449 MessageLoop::current()->Quit();
450 }
451 }
452
453 } // namespace IPC
OLDNEW
« no previous file with comments | « chrome/common/ipc_sync_channel.h ('k') | chrome/common/ipc_sync_channel_unittest.cc » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698