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

Side by Side Diff: ipc/ipc_sync_channel.cc

Issue 2067233002: Revert of Use Mojo pipes to signal sync IPC events (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@master
Patch Set: Created 4 years, 6 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 | « ipc/ipc_sync_channel.h ('k') | ipc/ipc_sync_message.h » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
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 #include "ipc/ipc_sync_channel.h" 5 #include "ipc/ipc_sync_channel.h"
6 6
7 #include <stddef.h> 7 #include <stddef.h>
8 #include <stdint.h> 8 #include <stdint.h>
9 9
10 #include <utility> 10 #include <utility>
11 11
12 #include "base/bind.h" 12 #include "base/bind.h"
13 #include "base/lazy_instance.h" 13 #include "base/lazy_instance.h"
14 #include "base/location.h" 14 #include "base/location.h"
15 #include "base/logging.h" 15 #include "base/logging.h"
16 #include "base/macros.h"
17 #include "base/memory/ptr_util.h" 16 #include "base/memory/ptr_util.h"
18 #include "base/run_loop.h"
19 #include "base/synchronization/waitable_event.h" 17 #include "base/synchronization/waitable_event.h"
18 #include "base/synchronization/waitable_event_watcher.h"
20 #include "base/threading/thread_local.h" 19 #include "base/threading/thread_local.h"
21 #include "base/threading/thread_task_runner_handle.h" 20 #include "base/threading/thread_task_runner_handle.h"
22 #include "base/trace_event/trace_event.h" 21 #include "base/trace_event/trace_event.h"
23 #include "ipc/ipc_channel_factory.h" 22 #include "ipc/ipc_channel_factory.h"
24 #include "ipc/ipc_logging.h" 23 #include "ipc/ipc_logging.h"
25 #include "ipc/ipc_message_macros.h" 24 #include "ipc/ipc_message_macros.h"
26 #include "ipc/ipc_sync_message.h" 25 #include "ipc/ipc_sync_message.h"
27 #include "ipc/mojo_event.h"
28 #include "mojo/public/cpp/bindings/sync_handle_registry.h"
29 26
30 using base::TimeDelta; 27 using base::TimeDelta;
31 using base::TimeTicks; 28 using base::TimeTicks;
32 using base::WaitableEvent; 29 using base::WaitableEvent;
33 30
34 namespace IPC { 31 namespace IPC {
35
36 namespace {
37
38 // A lazy thread-local Mojo Event which is always signaled. Used to wake up the
39 // sync waiter when a SyncMessage requires the MessageLoop to be pumped while
40 // waiting for a reply. This object is created lazily and ref-counted so it can
41 // be cleaned up when no longer in use.
42 class PumpMessagesEvent {
43 public:
44 // Acquires the event for this thread. Creates a new instance if necessary.
45 static PumpMessagesEvent* Acquire() {
46 PumpMessagesEvent* pump_messages_event = g_event_.Pointer()->Get();
47 if (!pump_messages_event) {
48 pump_messages_event = new PumpMessagesEvent;
49 pump_messages_event->event_.Signal();
50 g_event_.Pointer()->Set(pump_messages_event);
51 }
52 pump_messages_event->ref_count_++;
53 return pump_messages_event;
54 }
55
56 // Releases a handle to this event. There must be a 1:1 correspondence between
57 // calls to Acquire() and calls to Release().
58 static void Release() {
59 PumpMessagesEvent* pump_messages_event = g_event_.Pointer()->Get();
60 DCHECK(pump_messages_event);
61 DCHECK_GT(pump_messages_event->ref_count_, 0);
62 pump_messages_event->ref_count_--;
63 if (!pump_messages_event->ref_count_) {
64 g_event_.Pointer()->Set(nullptr);
65 delete pump_messages_event;
66 }
67 }
68
69 const mojo::Handle& GetHandle() const { return event_.GetHandle(); }
70
71 private:
72 PumpMessagesEvent() {}
73 ~PumpMessagesEvent() {}
74
75 int ref_count_ = 0;
76 MojoEvent event_;
77
78 static base::LazyInstance<base::ThreadLocalPointer<PumpMessagesEvent>>
79 g_event_;
80
81 DISALLOW_COPY_AND_ASSIGN(PumpMessagesEvent);
82 };
83
84 // A generic callback used when watching handles synchronously. Sets |*signal|
85 // to true. Also sets |*error| to true in case of an error.
86 void OnSyncHandleReady(bool* signal, bool* error, MojoResult result) {
87 *signal = true;
88 *error = result != MOJO_RESULT_OK;
89 }
90
91 // A ReadyCallback for use with mojo::Watcher. Ignores the result (DCHECKs, but
92 // is only used in cases where failure should be impossible) and runs
93 // |callback|.
94 void RunOnHandleReady(const base::Closure& callback, MojoResult result) {
95 DCHECK(result == MOJO_RESULT_OK || result == MOJO_RESULT_ABORTED);
96 if (result == MOJO_RESULT_OK)
97 callback.Run();
98 }
99
100 } // namespace
101
102 base::LazyInstance<base::ThreadLocalPointer<PumpMessagesEvent>>
103 PumpMessagesEvent::g_event_ = LAZY_INSTANCE_INITIALIZER;
104
105 // When we're blocked in a Send(), we need to process incoming synchronous 32 // When we're blocked in a Send(), we need to process incoming synchronous
106 // messages right away because it could be blocking our reply (either 33 // messages right away because it could be blocking our reply (either
107 // directly from the same object we're calling, or indirectly through one or 34 // directly from the same object we're calling, or indirectly through one or
108 // more other channels). That means that in SyncContext's OnMessageReceived, 35 // more other channels). That means that in SyncContext's OnMessageReceived,
109 // we need to process sync message right away if we're blocked. However a 36 // we need to process sync message right away if we're blocked. However a
110 // simple check isn't sufficient, because the listener thread can be in the 37 // simple check isn't sufficient, because the listener thread can be in the
111 // process of calling Send. 38 // process of calling Send.
112 // To work around this, when SyncChannel filters a sync message, it sets 39 // To work around this, when SyncChannel filters a sync message, it sets
113 // an event that the listener thread waits on during its Send() call. This 40 // an event that the listener thread waits on during its Send() call. This
114 // allows us to dispatch incoming sync messages when blocked. The race 41 // allows us to dispatch incoming sync messages when blocked. The race
(...skipping 106 matching lines...) Expand 10 before | Expand all | Expand 10 after
221 iter++; 148 iter++;
222 } 149 }
223 } 150 }
224 151
225 if (--listener_count_ == 0) { 152 if (--listener_count_ == 0) {
226 DCHECK(lazy_tls_ptr_.Pointer()->Get()); 153 DCHECK(lazy_tls_ptr_.Pointer()->Get());
227 lazy_tls_ptr_.Pointer()->Set(NULL); 154 lazy_tls_ptr_.Pointer()->Set(NULL);
228 } 155 }
229 } 156 }
230 157
231 MojoEvent* dispatch_event() { return &dispatch_event_; } 158 WaitableEvent* dispatch_event() { return &dispatch_event_; }
232 base::SingleThreadTaskRunner* listener_task_runner() { 159 base::SingleThreadTaskRunner* listener_task_runner() {
233 return listener_task_runner_.get(); 160 return listener_task_runner_.get();
234 } 161 }
235 162
236 // Holds a pointer to the per-thread ReceivedSyncMsgQueue object. 163 // Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
237 static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> > 164 static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> >
238 lazy_tls_ptr_; 165 lazy_tls_ptr_;
239 166
240 // Called on the ipc thread to check if we can unblock any current Send() 167 // Called on the ipc thread to check if we can unblock any current Send()
241 // calls based on a queued reply. 168 // calls based on a queued reply.
242 void DispatchReplies() { 169 void DispatchReplies() {
243 for (size_t i = 0; i < received_replies_.size(); ++i) { 170 for (size_t i = 0; i < received_replies_.size(); ++i) {
244 Message* message = received_replies_[i].message; 171 Message* message = received_replies_[i].message;
245 if (received_replies_[i].context->TryToUnblockListener(message)) { 172 if (received_replies_[i].context->TryToUnblockListener(message)) {
246 delete message; 173 delete message;
247 received_replies_.erase(received_replies_.begin() + i); 174 received_replies_.erase(received_replies_.begin() + i);
248 return; 175 return;
249 } 176 }
250 } 177 }
251 } 178 }
252 179
253 mojo::Watcher* top_send_done_watcher() { 180 base::WaitableEventWatcher* top_send_done_watcher() {
254 return top_send_done_watcher_; 181 return top_send_done_watcher_;
255 } 182 }
256 183
257 void set_top_send_done_watcher(mojo::Watcher* watcher) { 184 void set_top_send_done_watcher(base::WaitableEventWatcher* watcher) {
258 top_send_done_watcher_ = watcher; 185 top_send_done_watcher_ = watcher;
259 } 186 }
260 187
261 private: 188 private:
262 friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>; 189 friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>;
263 190
264 // See the comment in SyncChannel::SyncChannel for why this event is created 191 // See the comment in SyncChannel::SyncChannel for why this event is created
265 // as manual reset. 192 // as manual reset.
266 ReceivedSyncMsgQueue() 193 ReceivedSyncMsgQueue()
267 : message_queue_version_(0), 194 : message_queue_version_(0),
195 dispatch_event_(base::WaitableEvent::ResetPolicy::MANUAL,
196 base::WaitableEvent::InitialState::NOT_SIGNALED),
268 listener_task_runner_(base::ThreadTaskRunnerHandle::Get()), 197 listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
269 task_pending_(false), 198 task_pending_(false),
270 listener_count_(0), 199 listener_count_(0),
271 top_send_done_watcher_(NULL) {} 200 top_send_done_watcher_(NULL) {}
272 201
273 ~ReceivedSyncMsgQueue() {} 202 ~ReceivedSyncMsgQueue() {}
274 203
275 // Holds information about a queued synchronous message or reply. 204 // Holds information about a queued synchronous message or reply.
276 struct QueuedMessage { 205 struct QueuedMessage {
277 QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { } 206 QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
278 Message* message; 207 Message* message;
279 scoped_refptr<SyncChannel::SyncContext> context; 208 scoped_refptr<SyncChannel::SyncContext> context;
280 }; 209 };
281 210
282 typedef std::list<QueuedMessage> SyncMessageQueue; 211 typedef std::list<QueuedMessage> SyncMessageQueue;
283 SyncMessageQueue message_queue_; 212 SyncMessageQueue message_queue_;
284 uint32_t message_queue_version_; // Used to signal DispatchMessages to rescan 213 uint32_t message_queue_version_; // Used to signal DispatchMessages to rescan
285 214
286 std::vector<QueuedMessage> received_replies_; 215 std::vector<QueuedMessage> received_replies_;
287 216
288 // Signaled when we get a synchronous message that we must respond to, as the 217 // Set when we got a synchronous message that we must respond to as the
289 // sender needs its reply before it can reply to our original synchronous 218 // sender needs its reply before it can reply to our original synchronous
290 // message. 219 // message.
291 MojoEvent dispatch_event_; 220 WaitableEvent dispatch_event_;
292 scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_; 221 scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
293 base::Lock message_lock_; 222 base::Lock message_lock_;
294 bool task_pending_; 223 bool task_pending_;
295 int listener_count_; 224 int listener_count_;
296 225
297 // The current send done handle watcher for this thread. Used to maintain 226 // The current send done event watcher for this thread. Used to maintain
298 // a thread-local stack of send done watchers to ensure that nested sync 227 // a local global stack of send done watchers to ensure that nested sync
299 // message loops complete correctly. 228 // message loops complete correctly.
300 mojo::Watcher* top_send_done_watcher_; 229 base::WaitableEventWatcher* top_send_done_watcher_;
301 }; 230 };
302 231
303 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> > 232 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> >
304 SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_ = 233 SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_ =
305 LAZY_INSTANCE_INITIALIZER; 234 LAZY_INSTANCE_INITIALIZER;
306 235
307 SyncChannel::SyncContext::SyncContext( 236 SyncChannel::SyncContext::SyncContext(
308 Listener* listener, 237 Listener* listener,
309 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 238 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
310 WaitableEvent* shutdown_event) 239 WaitableEvent* shutdown_event)
(...skipping 15 matching lines...) Expand all
326 // Create the tracking information for this message. This object is stored 255 // Create the tracking information for this message. This object is stored
327 // by value since all members are pointers that are cheap to copy. These 256 // by value since all members are pointers that are cheap to copy. These
328 // pointers are cleaned up in the Pop() function. 257 // pointers are cleaned up in the Pop() function.
329 // 258 //
330 // The event is created as manual reset because in between Signal and 259 // The event is created as manual reset because in between Signal and
331 // OnObjectSignalled, another Send can happen which would stop the watcher 260 // OnObjectSignalled, another Send can happen which would stop the watcher
332 // from being called. The event would get watched later, when the nested 261 // from being called. The event would get watched later, when the nested
333 // Send completes, so the event will need to remain set. 262 // Send completes, so the event will need to remain set.
334 PendingSyncMsg pending( 263 PendingSyncMsg pending(
335 SyncMessage::GetMessageId(*sync_msg), sync_msg->GetReplyDeserializer(), 264 SyncMessage::GetMessageId(*sync_msg), sync_msg->GetReplyDeserializer(),
336 new MojoEvent); 265 new WaitableEvent(base::WaitableEvent::ResetPolicy::MANUAL,
266 base::WaitableEvent::InitialState::NOT_SIGNALED));
337 base::AutoLock auto_lock(deserializers_lock_); 267 base::AutoLock auto_lock(deserializers_lock_);
338 deserializers_.push_back(pending); 268 deserializers_.push_back(pending);
339 } 269 }
340 270
341 bool SyncChannel::SyncContext::Pop() { 271 bool SyncChannel::SyncContext::Pop() {
342 bool result; 272 bool result;
343 { 273 {
344 base::AutoLock auto_lock(deserializers_lock_); 274 base::AutoLock auto_lock(deserializers_lock_);
345 PendingSyncMsg msg = deserializers_.back(); 275 PendingSyncMsg msg = deserializers_.back();
346 delete msg.deserializer; 276 delete msg.deserializer;
347 delete msg.done_event; 277 delete msg.done_event;
348 msg.done_event = nullptr; 278 msg.done_event = NULL;
349 deserializers_.pop_back(); 279 deserializers_.pop_back();
350 result = msg.send_result; 280 result = msg.send_result;
351 } 281 }
352 282
353 // We got a reply to a synchronous Send() call that's blocking the listener 283 // We got a reply to a synchronous Send() call that's blocking the listener
354 // thread. However, further down the call stack there could be another 284 // thread. However, further down the call stack there could be another
355 // blocking Send() call, whose reply we received after we made this last 285 // blocking Send() call, whose reply we received after we made this last
356 // Send() call. So check if we have any queued replies available that 286 // Send() call. So check if we have any queued replies available that
357 // can now unblock the listener thread. 287 // can now unblock the listener thread.
358 ipc_task_runner()->PostTask( 288 ipc_task_runner()->PostTask(
359 FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchReplies, 289 FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchReplies,
360 received_sync_msgs_.get())); 290 received_sync_msgs_.get()));
361 291
362 return result; 292 return result;
363 } 293 }
364 294
365 MojoEvent* SyncChannel::SyncContext::GetSendDoneEvent() { 295 WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
366 base::AutoLock auto_lock(deserializers_lock_); 296 base::AutoLock auto_lock(deserializers_lock_);
367 return deserializers_.back().done_event; 297 return deserializers_.back().done_event;
368 } 298 }
369 299
370 MojoEvent* SyncChannel::SyncContext::GetDispatchEvent() { 300 WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
371 return received_sync_msgs_->dispatch_event(); 301 return received_sync_msgs_->dispatch_event();
372 } 302 }
373 303
374 void SyncChannel::SyncContext::DispatchMessages() { 304 void SyncChannel::SyncContext::DispatchMessages() {
375 received_sync_msgs_->DispatchMessages(this); 305 received_sync_msgs_->DispatchMessages(this);
376 } 306 }
377 307
378 bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) { 308 bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
379 base::AutoLock auto_lock(deserializers_lock_); 309 base::AutoLock auto_lock(deserializers_lock_);
380 if (deserializers_.empty() || 310 if (deserializers_.empty() ||
381 !SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) { 311 !SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
382 return false; 312 return false;
383 } 313 }
384 314
385 if (!msg->is_reply_error()) { 315 if (!msg->is_reply_error()) {
386 bool send_result = deserializers_.back().deserializer-> 316 bool send_result = deserializers_.back().deserializer->
387 SerializeOutputParameters(*msg); 317 SerializeOutputParameters(*msg);
388 deserializers_.back().send_result = send_result; 318 deserializers_.back().send_result = send_result;
389 DVLOG_IF(1, !send_result) << "Couldn't deserialize reply message"; 319 DVLOG_IF(1, !send_result) << "Couldn't deserialize reply message";
390 } else { 320 } else {
391 DVLOG(1) << "Received error reply"; 321 DVLOG(1) << "Received error reply";
392 } 322 }
393 323
394 MojoEvent* done_event = deserializers_.back().done_event; 324 base::WaitableEvent* done_event = deserializers_.back().done_event;
395 TRACE_EVENT_FLOW_BEGIN0( 325 TRACE_EVENT_FLOW_BEGIN0(
396 TRACE_DISABLED_BY_DEFAULT("ipc.flow"), 326 TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
397 "SyncChannel::SyncContext::TryToUnblockListener", done_event); 327 "SyncChannel::SyncContext::TryToUnblockListener", done_event);
398 328
399 done_event->Signal(); 329 done_event->Signal();
400 330
401 return true; 331 return true;
402 } 332 }
403 333
404 void SyncChannel::SyncContext::Clear() { 334 void SyncChannel::SyncContext::Clear() {
(...skipping 25 matching lines...) Expand all
430 360
431 void SyncChannel::SyncContext::OnChannelError() { 361 void SyncChannel::SyncContext::OnChannelError() {
432 CancelPendingSends(); 362 CancelPendingSends();
433 shutdown_watcher_.StopWatching(); 363 shutdown_watcher_.StopWatching();
434 Context::OnChannelError(); 364 Context::OnChannelError();
435 } 365 }
436 366
437 void SyncChannel::SyncContext::OnChannelOpened() { 367 void SyncChannel::SyncContext::OnChannelOpened() {
438 shutdown_watcher_.StartWatching( 368 shutdown_watcher_.StartWatching(
439 shutdown_event_, 369 shutdown_event_,
440 base::Bind(&SyncChannel::SyncContext::OnShutdownEventSignaled, 370 base::Bind(&SyncChannel::SyncContext::OnWaitableEventSignaled,
441 base::Unretained(this))); 371 base::Unretained(this)));
442 Context::OnChannelOpened(); 372 Context::OnChannelOpened();
443 } 373 }
444 374
445 void SyncChannel::SyncContext::OnChannelClosed() { 375 void SyncChannel::SyncContext::OnChannelClosed() {
446 CancelPendingSends(); 376 CancelPendingSends();
447 shutdown_watcher_.StopWatching(); 377 shutdown_watcher_.StopWatching();
448 Context::OnChannelClosed(); 378 Context::OnChannelClosed();
449 } 379 }
450 380
451 void SyncChannel::SyncContext::CancelPendingSends() { 381 void SyncChannel::SyncContext::CancelPendingSends() {
452 base::AutoLock auto_lock(deserializers_lock_); 382 base::AutoLock auto_lock(deserializers_lock_);
453 PendingSyncMessageQueue::iterator iter; 383 PendingSyncMessageQueue::iterator iter;
454 DVLOG(1) << "Canceling pending sends"; 384 DVLOG(1) << "Canceling pending sends";
455 for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) { 385 for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
456 TRACE_EVENT_FLOW_BEGIN0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"), 386 TRACE_EVENT_FLOW_BEGIN0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
457 "SyncChannel::SyncContext::CancelPendingSends", 387 "SyncChannel::SyncContext::CancelPendingSends",
458 iter->done_event); 388 iter->done_event);
459 iter->done_event->Signal(); 389 iter->done_event->Signal();
460 } 390 }
461 } 391 }
462 392
463 void SyncChannel::SyncContext::OnShutdownEventSignaled(WaitableEvent* event) { 393 void SyncChannel::SyncContext::OnWaitableEventSignaled(WaitableEvent* event) {
464 DCHECK_EQ(event, shutdown_event_); 394 if (event == shutdown_event_) {
395 // Process shut down before we can get a reply to a synchronous message.
396 // Cancel pending Send calls, which will end up setting the send done event.
397 CancelPendingSends();
398 } else {
399 // We got the reply, timed out or the process shutdown.
400 DCHECK_EQ(GetSendDoneEvent(), event);
401 base::MessageLoop::current()->QuitNow();
402 }
403 }
465 404
466 // Process shut down before we can get a reply to a synchronous message. 405 base::WaitableEventWatcher::EventCallback
467 // Cancel pending Send calls, which will end up setting the send done event. 406 SyncChannel::SyncContext::MakeWaitableEventCallback() {
468 CancelPendingSends(); 407 return base::Bind(&SyncChannel::SyncContext::OnWaitableEventSignaled, this);
469 } 408 }
470 409
471 // static 410 // static
472 std::unique_ptr<SyncChannel> SyncChannel::Create( 411 std::unique_ptr<SyncChannel> SyncChannel::Create(
473 const IPC::ChannelHandle& channel_handle, 412 const IPC::ChannelHandle& channel_handle,
474 Channel::Mode mode, 413 Channel::Mode mode,
475 Listener* listener, 414 Listener* listener,
476 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 415 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
477 bool create_pipe_now, 416 bool create_pipe_now,
478 base::WaitableEvent* shutdown_event) { 417 base::WaitableEvent* shutdown_event) {
(...skipping 23 matching lines...) Expand all
502 WaitableEvent* shutdown_event) { 441 WaitableEvent* shutdown_event) {
503 return base::WrapUnique( 442 return base::WrapUnique(
504 new SyncChannel(listener, ipc_task_runner, shutdown_event)); 443 new SyncChannel(listener, ipc_task_runner, shutdown_event));
505 } 444 }
506 445
507 SyncChannel::SyncChannel( 446 SyncChannel::SyncChannel(
508 Listener* listener, 447 Listener* listener,
509 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 448 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
510 WaitableEvent* shutdown_event) 449 WaitableEvent* shutdown_event)
511 : ChannelProxy(new SyncContext(listener, ipc_task_runner, shutdown_event)) { 450 : ChannelProxy(new SyncContext(listener, ipc_task_runner, shutdown_event)) {
512 // Keep a thread-local PumpMessagesEvent alive at least as long as any
513 // SyncChannel exists. This is balanced in the SyncChannel destructor below.
514 PumpMessagesEvent::Acquire();
515
516 // The current (listener) thread must be distinct from the IPC thread, or else 451 // The current (listener) thread must be distinct from the IPC thread, or else
517 // sending synchronous messages will deadlock. 452 // sending synchronous messages will deadlock.
518 DCHECK_NE(ipc_task_runner.get(), base::ThreadTaskRunnerHandle::Get().get()); 453 DCHECK_NE(ipc_task_runner.get(), base::ThreadTaskRunnerHandle::Get().get());
519 StartWatching(); 454 StartWatching();
520 } 455 }
521 456
522 SyncChannel::~SyncChannel() { 457 SyncChannel::~SyncChannel() {
523 PumpMessagesEvent::Release();
524 } 458 }
525 459
526 void SyncChannel::SetRestrictDispatchChannelGroup(int group) { 460 void SyncChannel::SetRestrictDispatchChannelGroup(int group) {
527 sync_context()->set_restrict_dispatch_group(group); 461 sync_context()->set_restrict_dispatch_group(group);
528 } 462 }
529 463
530 scoped_refptr<SyncMessageFilter> SyncChannel::CreateSyncMessageFilter() { 464 scoped_refptr<SyncMessageFilter> SyncChannel::CreateSyncMessageFilter() {
531 scoped_refptr<SyncMessageFilter> filter = new SyncMessageFilter( 465 scoped_refptr<SyncMessageFilter> filter = new SyncMessageFilter(
532 sync_context()->shutdown_event(), 466 sync_context()->shutdown_event(),
533 sync_context()->IsChannelSendThreadSafe()); 467 sync_context()->IsChannelSendThreadSafe());
(...skipping 20 matching lines...) Expand all
554 488
555 // *this* might get deleted in WaitForReply. 489 // *this* might get deleted in WaitForReply.
556 scoped_refptr<SyncContext> context(sync_context()); 490 scoped_refptr<SyncContext> context(sync_context());
557 if (context->shutdown_event()->IsSignaled()) { 491 if (context->shutdown_event()->IsSignaled()) {
558 DVLOG(1) << "shutdown event is signaled"; 492 DVLOG(1) << "shutdown event is signaled";
559 delete message; 493 delete message;
560 return false; 494 return false;
561 } 495 }
562 496
563 SyncMessage* sync_msg = static_cast<SyncMessage*>(message); 497 SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
564 bool pump_messages = sync_msg->ShouldPumpMessages();
565 context->Push(sync_msg); 498 context->Push(sync_msg);
499 WaitableEvent* pump_messages_event = sync_msg->pump_messages_event();
566 500
567 ChannelProxy::Send(message); 501 ChannelProxy::Send(message);
568 502
569 // Wait for reply, or for any other incoming synchronous messages. 503 // Wait for reply, or for any other incoming synchronous messages.
570 // *this* might get deleted, so only call static functions at this point. 504 // *this* might get deleted, so only call static functions at this point.
571 WaitForReply(context.get(), pump_messages); 505 WaitForReply(context.get(), pump_messages_event);
572 506
573 TRACE_EVENT_FLOW_END0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"), 507 TRACE_EVENT_FLOW_END0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
574 "SyncChannel::Send", context->GetSendDoneEvent()); 508 "SyncChannel::Send", context->GetSendDoneEvent());
575 509
576 return context->Pop(); 510 return context->Pop();
577 } 511 }
578 512
579 void SyncChannel::WaitForReply(SyncContext* context, bool pump_messages) { 513 void SyncChannel::WaitForReply(
514 SyncContext* context, WaitableEvent* pump_messages_event) {
580 context->DispatchMessages(); 515 context->DispatchMessages();
516 while (true) {
517 WaitableEvent* objects[] = {
518 context->GetDispatchEvent(),
519 context->GetSendDoneEvent(),
520 pump_messages_event
521 };
581 522
582 PumpMessagesEvent* pump_messages_event = nullptr; 523 unsigned count = pump_messages_event ? 3: 2;
583 if (pump_messages) 524 size_t result = WaitableEvent::WaitMany(objects, count);
584 pump_messages_event = PumpMessagesEvent::Acquire(); 525 if (result == 0 /* dispatch event */) {
585
586 scoped_refptr<mojo::SyncHandleRegistry> registry =
587 mojo::SyncHandleRegistry::current();
588
589 while (true) {
590 bool dispatch = false;
591 bool send_done = false;
592 bool should_pump_messages = false;
593 bool error = false;
594 registry->RegisterHandle(context->GetDispatchEvent()->GetHandle(),
595 MOJO_HANDLE_SIGNAL_READABLE,
596 base::Bind(&OnSyncHandleReady, &dispatch, &error));
597 registry->RegisterHandle(
598 context->GetSendDoneEvent()->GetHandle(),
599 MOJO_HANDLE_SIGNAL_READABLE,
600 base::Bind(&OnSyncHandleReady, &send_done, &error));
601 if (pump_messages_event) {
602 registry->RegisterHandle(
603 pump_messages_event->GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
604 base::Bind(&OnSyncHandleReady, &should_pump_messages, &error));
605 }
606
607 const bool* stop_flags[] = { &dispatch, &send_done, &should_pump_messages };
608 bool result = registry->WatchAllHandles(stop_flags, 3);
609 DCHECK(result);
610 DCHECK(!error);
611
612 registry->UnregisterHandle(context->GetDispatchEvent()->GetHandle());
613 registry->UnregisterHandle(context->GetSendDoneEvent()->GetHandle());
614 if (pump_messages_event)
615 registry->UnregisterHandle(pump_messages_event->GetHandle());
616
617 if (dispatch) {
618 // We're waiting for a reply, but we received a blocking synchronous 526 // We're waiting for a reply, but we received a blocking synchronous
619 // call. We must process it or otherwise a deadlock might occur. 527 // call. We must process it or otherwise a deadlock might occur.
620 context->GetDispatchEvent()->Reset(); 528 context->GetDispatchEvent()->Reset();
621 context->DispatchMessages(); 529 context->DispatchMessages();
622 continue; 530 continue;
623 } 531 }
624 532
625 DCHECK(send_done || should_pump_messages); 533 if (result == 2 /* pump_messages_event */)
626
627 if (should_pump_messages)
628 WaitForReplyWithNestedMessageLoop(context); // Run a nested message loop. 534 WaitForReplyWithNestedMessageLoop(context); // Run a nested message loop.
629 535
630 break; 536 break;
631 } 537 }
632
633 if (pump_messages_event)
634 PumpMessagesEvent::Release();
635 } 538 }
636 539
637 void SyncChannel::WaitForReplyWithNestedMessageLoop(SyncContext* context) { 540 void SyncChannel::WaitForReplyWithNestedMessageLoop(SyncContext* context) {
638 mojo::Watcher send_done_watcher; 541 base::WaitableEventWatcher send_done_watcher;
639 542
640 ReceivedSyncMsgQueue* sync_msg_queue = context->received_sync_msgs(); 543 ReceivedSyncMsgQueue* sync_msg_queue = context->received_sync_msgs();
641 DCHECK_NE(sync_msg_queue, nullptr); 544 DCHECK(sync_msg_queue != NULL);
642 545
643 mojo::Watcher* old_watcher = sync_msg_queue->top_send_done_watcher(); 546 base::WaitableEventWatcher* old_send_done_event_watcher =
644 mojo::Handle old_handle(mojo::kInvalidHandleValue); 547 sync_msg_queue->top_send_done_watcher();
645 mojo::Watcher::ReadyCallback old_callback;
646 548
647 // Maintain a thread-local stack of watchers to ensure nested calls complete 549 base::WaitableEventWatcher::EventCallback old_callback;
648 // in the correct sequence, i.e. the outermost call completes first, etc. 550 base::WaitableEvent* old_event = NULL;
649 if (old_watcher) { 551
650 old_callback = old_watcher->ready_callback(); 552 // Maintain a local global stack of send done delegates to ensure that
651 old_handle = old_watcher->handle(); 553 // nested sync calls complete in the correct sequence, i.e. the
652 old_watcher->Cancel(); 554 // outermost call completes first, etc.
555 if (old_send_done_event_watcher) {
556 old_callback = old_send_done_event_watcher->callback();
557 old_event = old_send_done_event_watcher->GetWatchedEvent();
558 old_send_done_event_watcher->StopWatching();
653 } 559 }
654 560
655 sync_msg_queue->set_top_send_done_watcher(&send_done_watcher); 561 sync_msg_queue->set_top_send_done_watcher(&send_done_watcher);
656 562
563 send_done_watcher.StartWatching(context->GetSendDoneEvent(),
564 context->MakeWaitableEventCallback());
565
657 { 566 {
658 base::RunLoop nested_loop;
659 send_done_watcher.Start(
660 context->GetSendDoneEvent()->GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
661 base::Bind(&RunOnHandleReady, nested_loop.QuitClosure()));
662
663 base::MessageLoop::ScopedNestableTaskAllower allow( 567 base::MessageLoop::ScopedNestableTaskAllower allow(
664 base::MessageLoop::current()); 568 base::MessageLoop::current());
665 nested_loop.Run(); 569 base::MessageLoop::current()->Run();
666 send_done_watcher.Cancel();
667 } 570 }
668 571
669 sync_msg_queue->set_top_send_done_watcher(old_watcher); 572 sync_msg_queue->set_top_send_done_watcher(old_send_done_event_watcher);
670 if (old_watcher) 573 if (old_send_done_event_watcher && old_event) {
671 old_watcher->Start(old_handle, MOJO_HANDLE_SIGNAL_READABLE, old_callback); 574 old_send_done_event_watcher->StartWatching(old_event, old_callback);
575 }
672 } 576 }
673 577
674 void SyncChannel::OnDispatchHandleReady(MojoResult result) { 578 void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) {
675 DCHECK(result == MOJO_RESULT_OK || result == MOJO_RESULT_ABORTED); 579 DCHECK(event == sync_context()->GetDispatchEvent());
676 if (result == MOJO_RESULT_OK) { 580 // The call to DispatchMessages might delete this object, so reregister
677 sync_context()->GetDispatchEvent()->Reset(); 581 // the object watcher first.
678 sync_context()->DispatchMessages(); 582 event->Reset();
679 } 583 dispatch_watcher_.StartWatching(event, dispatch_watcher_callback_);
584 sync_context()->DispatchMessages();
680 } 585 }
681 586
682 void SyncChannel::StartWatching() { 587 void SyncChannel::StartWatching() {
683 // Ideally we only want to watch this object when running a nested message 588 // Ideally we only want to watch this object when running a nested message
684 // loop. However, we don't know when it exits if there's another nested 589 // loop. However, we don't know when it exits if there's another nested
685 // message loop running under it or not, so we wouldn't know whether to 590 // message loop running under it or not, so we wouldn't know whether to
686 // stop or keep watching. So we always watch it. 591 // stop or keep watching. So we always watch it, and create the event as
687 dispatch_watcher_.Start(sync_context()->GetDispatchEvent()->GetHandle(), 592 // manual reset since the object watcher might otherwise reset the event
688 MOJO_HANDLE_SIGNAL_READABLE, 593 // when we're doing a WaitMany.
689 base::Bind(&SyncChannel::OnDispatchHandleReady, 594 dispatch_watcher_callback_ =
690 base::Unretained(this))); 595 base::Bind(&SyncChannel::OnWaitableEventSignaled,
596 base::Unretained(this));
597 dispatch_watcher_.StartWatching(sync_context()->GetDispatchEvent(),
598 dispatch_watcher_callback_);
691 } 599 }
692 600
693 void SyncChannel::OnChannelInit() { 601 void SyncChannel::OnChannelInit() {
694 for (const auto& filter : pre_init_sync_message_filters_) { 602 for (const auto& filter : pre_init_sync_message_filters_) {
695 filter->set_is_channel_send_thread_safe( 603 filter->set_is_channel_send_thread_safe(
696 context()->IsChannelSendThreadSafe()); 604 context()->IsChannelSendThreadSafe());
697 } 605 }
698 pre_init_sync_message_filters_.clear(); 606 pre_init_sync_message_filters_.clear();
699 } 607 }
700 608
(...skipping 23 matching lines...) Expand all
724 TRACE_EVENT2("ipc", "SyncChannel::SendOnIPCThread", 632 TRACE_EVENT2("ipc", "SyncChannel::SendOnIPCThread",
725 "class", IPC_MESSAGE_ID_CLASS(message->type()), 633 "class", IPC_MESSAGE_ID_CLASS(message->type()),
726 "line", IPC_MESSAGE_ID_LINE(message->type())); 634 "line", IPC_MESSAGE_ID_LINE(message->type()));
727 #endif 635 #endif
728 if (!message->is_sync()) 636 if (!message->is_sync())
729 return ChannelProxy::SendOnIPCThread(std::move(message)); 637 return ChannelProxy::SendOnIPCThread(std::move(message));
730 return Send(message.release()); 638 return Send(message.release());
731 } 639 }
732 640
733 } // namespace IPC 641 } // namespace IPC
OLDNEW
« no previous file with comments | « ipc/ipc_sync_channel.h ('k') | ipc/ipc_sync_message.h » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698