ipc/ipc_sync_channel.cc

   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
   3 // found in the LICENSE file.
   4
   5 #include "ipc/ipc_sync_channel.h"
   6
   7 #include "base/bind.h"
   8 #include "base/lazy_instance.h"
   9 #include "base/location.h"
  10 #include "base/logging.h"
  11 #include "base/threading/thread_local.h"
  12 #include "base/synchronization/waitable_event.h"
  13 #include "base/synchronization/waitable_event_watcher.h"
  14 #include "ipc/ipc_sync_message.h"
  15
  16 using base::TimeDelta;
  17 using base::TimeTicks;
  18 using base::WaitableEvent;
  19
  20 namespace IPC {
  21 // When we're blocked in a Send(), we need to process incoming synchronous
  22 // messages right away because it could be blocking our reply (either
  23 // directly from the same object we're calling, or indirectly through one or
  24 // more other channels).  That means that in SyncContext's OnMessageReceived,
  25 // we need to process sync message right away if we're blocked.  However a
  26 // simple check isn't sufficient, because the listener thread can be in the
  27 // process of calling Send.
  28 // To work around this, when SyncChannel filters a sync message, it sets
  29 // an event that the listener thread waits on during its Send() call.  This
  30 // allows us to dispatch incoming sync messages when blocked.  The race
  31 // condition is handled because if Send is in the process of being called, it
  32 // will check the event.  In case the listener thread isn't sending a message,
  33 // we queue a task on the listener thread to dispatch the received messages.
  34 // The messages are stored in this queue object that's shared among all
  35 // SyncChannel objects on the same thread (since one object can receive a
  36 // sync message while another one is blocked).
  37
  38 class SyncChannel::ReceivedSyncMsgQueue :
  39     public base::RefCountedThreadSafe<ReceivedSyncMsgQueue> {
  40  public:
  41   // Returns the ReceivedSyncMsgQueue instance for this thread, creating one
  42   // if necessary.  Call RemoveContext on the same thread when done.
  43   static ReceivedSyncMsgQueue* AddContext() {
  44     // We want one ReceivedSyncMsgQueue per listener thread (i.e. since multiple
  45     // SyncChannel objects can block the same thread).
  46     ReceivedSyncMsgQueue* rv = lazy_tls_ptr_.Pointer()->Get();
  47     if (!rv) {
  48       rv = new ReceivedSyncMsgQueue();
  49       ReceivedSyncMsgQueue::lazy_tls_ptr_.Pointer()->Set(rv);
  50     }
  51     rv->listener_count_++;
  52     return rv;
  53   }
  54
  55   // Called on IPC thread when a synchronous message or reply arrives.
  56   void QueueMessage(const Message& msg, SyncChannel::SyncContext* context) {
  57     bool was_task_pending;
  58     {
  59       base::AutoLock auto_lock(message_lock_);
  60
  61       was_task_pending = task_pending_;
  62       task_pending_ = true;
  63
  64       // We set the event in case the listener thread is blocked (or is about
  65       // to). In case it's not, the PostTask dispatches the messages.
  66       message_queue_.push_back(QueuedMessage(new Message(msg), context));
  67       message_queue_version_++;
  68     }
  69
  70     dispatch_event_.Signal();
  71     if (!was_task_pending) {
  72       listener_message_loop_->PostTask(
  73           FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchMessagesTask,
  74                                 this, scoped_refptr<SyncContext>(context)));
  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(SyncContext* context) {
  85     {
  86       base::AutoLock auto_lock(message_lock_);
  87       task_pending_ = false;
  88     }
  89     context->DispatchMessages();
  90   }
  91
  92   void DispatchMessages(SyncContext* dispatching_context) {
  93     bool first_time = true;
  94     uint32 expected_version = 0;
  95     SyncMessageQueue::iterator it;
  96     while (true) {
  97       Message* message = NULL;
  98       scoped_refptr<SyncChannel::SyncContext> context;
  99       {
 100         base::AutoLock auto_lock(message_lock_);
 101         if (first_time || message_queue_version_ != expected_version) {
 102           it = message_queue_.begin();
 103           first_time = false;
 104         }
 105         for (; it != message_queue_.end(); it++) {
 106           if (!it->context->restrict_dispatch() ||
 107               it->context == dispatching_context) {
 108             message = it->message;
 109             context = it->context;
 110             it = message_queue_.erase(it);
 111             message_queue_version_++;
 112             expected_version = message_queue_version_;
 113             break;
 114           }
 115         }
 116       }
 117
 118       if (message == NULL)
 119         break;
 120       context->OnDispatchMessage(*message);
 121       delete message;
 122     }
 123   }
 124
 125   // SyncChannel calls this in its destructor.
 126   void RemoveContext(SyncContext* context) {
 127     base::AutoLock auto_lock(message_lock_);
 128
 129     SyncMessageQueue::iterator iter = message_queue_.begin();
 130     while (iter != message_queue_.end()) {
 131       if (iter->context == context) {
 132         delete iter->message;
 133         iter = message_queue_.erase(iter);
 134         message_queue_version_++;
 135       } else {
 136         iter++;
 137       }
 138     }
 139
 140     if (--listener_count_ == 0) {
 141       DCHECK(lazy_tls_ptr_.Pointer()->Get());
 142       lazy_tls_ptr_.Pointer()->Set(NULL);
 143     }
 144   }
 145
 146   WaitableEvent* dispatch_event() { return &dispatch_event_; }
 147   base::MessageLoopProxy* listener_message_loop() {
 148     return listener_message_loop_;
 149   }
 150
 151   // Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
 152   static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> >
 153       lazy_tls_ptr_;
 154
 155   // Called on the ipc thread to check if we can unblock any current Send()
 156   // calls based on a queued reply.
 157   void DispatchReplies() {
 158     for (size_t i = 0; i < received_replies_.size(); ++i) {
 159       Message* message = received_replies_[i].message;
 160       if (received_replies_[i].context->TryToUnblockListener(message)) {
 161         delete message;
 162         received_replies_.erase(received_replies_.begin() + i);
 163         return;
 164       }
 165     }
 166   }
 167
 168   base::WaitableEventWatcher* top_send_done_watcher() {
 169     return top_send_done_watcher_;
 170   }
 171
 172   void set_top_send_done_watcher(base::WaitableEventWatcher* watcher) {
 173     top_send_done_watcher_ = watcher;
 174   }
 175
 176  private:
 177   friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>;
 178
 179   // See the comment in SyncChannel::SyncChannel for why this event is created
 180   // as manual reset.
 181   ReceivedSyncMsgQueue() :
 182       message_queue_version_(0),
 183       dispatch_event_(true, false),
 184       listener_message_loop_(base::MessageLoopProxy::current()),
 185       task_pending_(false),
 186       listener_count_(0),
 187       top_send_done_watcher_(NULL) {
 188   }
 189
 190   ~ReceivedSyncMsgQueue() {}
 191
 192   // Holds information about a queued synchronous message or reply.
 193   struct QueuedMessage {
 194     QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
 195     Message* message;
 196     scoped_refptr<SyncChannel::SyncContext> context;
 197   };
 198
 199   typedef std::list<QueuedMessage> SyncMessageQueue;
 200   SyncMessageQueue message_queue_;
 201   uint32 message_queue_version_;  // Used to signal DispatchMessages to rescan
 202
 203   std::vector<QueuedMessage> received_replies_;
 204
 205   // Set when we got a synchronous message that we must respond to as the
 206   // sender needs its reply before it can reply to our original synchronous
 207   // message.
 208   WaitableEvent dispatch_event_;
 209   scoped_refptr<base::MessageLoopProxy> listener_message_loop_;
 210   base::Lock message_lock_;
 211   bool task_pending_;
 212   int listener_count_;
 213
 214   // The current send done event watcher for this thread. Used to maintain
 215   // a local global stack of send done watchers to ensure that nested sync
 216   // message loops complete correctly.
 217   base::WaitableEventWatcher* top_send_done_watcher_;
 218 };
 219
 220 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> >
 221     SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_ =
 222         LAZY_INSTANCE_INITIALIZER;
 223
 224 SyncChannel::SyncContext::SyncContext(
 225     Channel::Listener* listener,
 226     base::MessageLoopProxy* ipc_thread,
 227     WaitableEvent* shutdown_event)
 228     : ChannelProxy::Context(listener, ipc_thread),
 229       received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
 230       shutdown_event_(shutdown_event),
 231       restrict_dispatch_(false) {
 232 }
 233
 234 SyncChannel::SyncContext::~SyncContext() {
 235   while (!deserializers_.empty())
 236     Pop();
 237 }
 238
 239 // Adds information about an outgoing sync message to the context so that
 240 // we know how to deserialize the reply.  Returns a handle that's set when
 241 // the reply has arrived.
 242 void SyncChannel::SyncContext::Push(SyncMessage* sync_msg) {
 243   // Create the tracking information for this message. This object is stored
 244   // by value since all members are pointers that are cheap to copy. These
 245   // pointers are cleaned up in the Pop() function.
 246   //
 247   // The event is created as manual reset because in between Signal and
 248   // OnObjectSignalled, another Send can happen which would stop the watcher
 249   // from being called.  The event would get watched later, when the nested
 250   // Send completes, so the event will need to remain set.
 251   PendingSyncMsg pending(SyncMessage::GetMessageId(*sync_msg),
 252                          sync_msg->GetReplyDeserializer(),
 253                          new WaitableEvent(true, false));
 254   base::AutoLock auto_lock(deserializers_lock_);
 255   deserializers_.push_back(pending);
 256 }
 257
 258 bool SyncChannel::SyncContext::Pop() {
 259   bool result;
 260   {
 261     base::AutoLock auto_lock(deserializers_lock_);
 262     PendingSyncMsg msg = deserializers_.back();
 263     delete msg.deserializer;
 264     delete msg.done_event;
 265     msg.done_event = NULL;
 266     deserializers_.pop_back();
 267     result = msg.send_result;
 268   }
 269
 270   // We got a reply to a synchronous Send() call that's blocking the listener
 271   // thread.  However, further down the call stack there could be another
 272   // blocking Send() call, whose reply we received after we made this last
 273   // Send() call.  So check if we have any queued replies available that
 274   // can now unblock the listener thread.
 275   ipc_message_loop()->PostTask(
 276       FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchReplies,
 277                             received_sync_msgs_.get()));
 278
 279   return result;
 280 }
 281
 282 WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
 283   base::AutoLock auto_lock(deserializers_lock_);
 284   return deserializers_.back().done_event;
 285 }
 286
 287 WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
 288   return received_sync_msgs_->dispatch_event();
 289 }
 290
 291 void SyncChannel::SyncContext::DispatchMessages() {
 292   received_sync_msgs_->DispatchMessages(this);
 293 }
 294
 295 bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
 296   base::AutoLock auto_lock(deserializers_lock_);
 297   if (deserializers_.empty() ||
 298       !SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
 299     return false;
 300   }
 301
 302   if (!msg->is_reply_error()) {
 303     deserializers_.back().send_result = deserializers_.back().deserializer->
 304         SerializeOutputParameters(*msg);
 305   }
 306   deserializers_.back().done_event->Signal();
 307
 308   return true;
 309 }
 310
 311 void SyncChannel::SyncContext::Clear() {
 312   CancelPendingSends();
 313   received_sync_msgs_->RemoveContext(this);
 314   Context::Clear();
 315 }
 316
 317 bool SyncChannel::SyncContext::OnMessageReceived(const Message& msg) {
 318   // Give the filters a chance at processing this message.
 319   if (TryFilters(msg))
 320     return true;
 321
 322   if (TryToUnblockListener(&msg))
 323     return true;
 324
 325   if (msg.should_unblock()) {
 326     received_sync_msgs_->QueueMessage(msg, this);
 327     return true;
 328   }
 329
 330   if (msg.is_reply()) {
 331     received_sync_msgs_->QueueReply(msg, this);
 332     return true;
 333   }
 334
 335   return Context::OnMessageReceivedNoFilter(msg);
 336 }
 337
 338 void SyncChannel::SyncContext::OnChannelError() {
 339   CancelPendingSends();
 340   shutdown_watcher_.StopWatching();
 341   Context::OnChannelError();
 342 }
 343
 344 void SyncChannel::SyncContext::OnChannelOpened() {
 345   shutdown_watcher_.StartWatching(shutdown_event_, this);
 346   Context::OnChannelOpened();
 347 }
 348
 349 void SyncChannel::SyncContext::OnChannelClosed() {
 350   CancelPendingSends();
 351   shutdown_watcher_.StopWatching();
 352   Context::OnChannelClosed();
 353 }
 354
 355 void SyncChannel::SyncContext::OnSendTimeout(int message_id) {
 356   base::AutoLock auto_lock(deserializers_lock_);
 357   PendingSyncMessageQueue::iterator iter;
 358   for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
 359     if (iter->id == message_id) {
 360       iter->done_event->Signal();
 361       break;
 362     }
 363   }
 364 }
 365
 366 void SyncChannel::SyncContext::CancelPendingSends() {
 367   base::AutoLock auto_lock(deserializers_lock_);
 368   PendingSyncMessageQueue::iterator iter;
 369   for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++)
 370     iter->done_event->Signal();
 371 }
 372
 373 void SyncChannel::SyncContext::OnWaitableEventSignaled(WaitableEvent* event) {
 374   if (event == shutdown_event_) {
 375     // Process shut down before we can get a reply to a synchronous message.
 376     // Cancel pending Send calls, which will end up setting the send done event.
 377     CancelPendingSends();
 378   } else {
 379     // We got the reply, timed out or the process shutdown.
 380     DCHECK_EQ(GetSendDoneEvent(), event);
 381     MessageLoop::current()->QuitNow();
 382   }
 383 }
 384
 385
 386 SyncChannel::SyncChannel(
 387     const IPC::ChannelHandle& channel_handle,
 388     Channel::Mode mode,
 389     Channel::Listener* listener,
 390     base::MessageLoopProxy* ipc_message_loop,
 391     bool create_pipe_now,
 392     WaitableEvent* shutdown_event)
 393     : ChannelProxy(new SyncContext(listener, ipc_message_loop, shutdown_event)),
 394       sync_messages_with_no_timeout_allowed_(true) {
 395   ChannelProxy::Init(channel_handle, mode, create_pipe_now);
 396   StartWatching();
 397 }
 398
 399 SyncChannel::SyncChannel(
 400     Channel::Listener* listener,
 401     base::MessageLoopProxy* ipc_message_loop,
 402     WaitableEvent* shutdown_event)
 403     : ChannelProxy(new SyncContext(listener, ipc_message_loop, shutdown_event)),
 404       sync_messages_with_no_timeout_allowed_(true) {
 405   StartWatching();
 406 }
 407
 408 SyncChannel::~SyncChannel() {
 409 }
 410
 411 void SyncChannel::SetRestrictDispatchToSameChannel(bool value) {
 412   sync_context()->set_restrict_dispatch(value);
 413 }
 414
 415 bool SyncChannel::Send(Message* message) {
 416   return SendWithTimeout(message, base::kNoTimeout);
 417 }
 418
 419 bool SyncChannel::SendWithTimeout(Message* message, int timeout_ms) {
 420   if (!message->is_sync()) {
 421     ChannelProxy::Send(message);
 422     return true;
 423   }
 424
 425   // *this* might get deleted in WaitForReply.
 426   scoped_refptr<SyncContext> context(sync_context());
 427   if (context->shutdown_event()->IsSignaled()) {
 428     delete message;
 429     return false;
 430   }
 431
 432   DCHECK(sync_messages_with_no_timeout_allowed_ ||
 433          timeout_ms != base::kNoTimeout);
 434   SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
 435   context->Push(sync_msg);
 436   int message_id = SyncMessage::GetMessageId(*sync_msg);
 437   WaitableEvent* pump_messages_event = sync_msg->pump_messages_event();
 438
 439   ChannelProxy::Send(message);
 440
 441   if (timeout_ms != base::kNoTimeout) {
 442     // We use the sync message id so that when a message times out, we don't
 443     // confuse it with another send that is either above/below this Send in
 444     // the call stack.
 445     context->ipc_message_loop()->PostDelayedTask(
 446         FROM_HERE,
 447         base::Bind(&SyncContext::OnSendTimeout, context.get(), message_id),
 448         timeout_ms);
 449   }
 450
 451   // Wait for reply, or for any other incoming synchronous messages.
 452   // *this* might get deleted, so only call static functions at this point.
 453   WaitForReply(context, pump_messages_event);
 454
 455   return context->Pop();
 456 }
 457
 458 void SyncChannel::WaitForReply(
 459     SyncContext* context, WaitableEvent* pump_messages_event) {
 460   context->DispatchMessages();
 461   while (true) {
 462     WaitableEvent* objects[] = {
 463       context->GetDispatchEvent(),
 464       context->GetSendDoneEvent(),
 465       pump_messages_event
 466     };
 467
 468     unsigned count = pump_messages_event ? 3: 2;
 469     size_t result = WaitableEvent::WaitMany(objects, count);
 470     if (result == 0 /* dispatch event */) {
 471       // We're waiting for a reply, but we received a blocking synchronous
 472       // call.  We must process it or otherwise a deadlock might occur.
 473       context->GetDispatchEvent()->Reset();
 474       context->DispatchMessages();
 475       continue;
 476     }
 477
 478     if (result == 2 /* pump_messages_event */)
 479       WaitForReplyWithNestedMessageLoop(context);  // Run a nested message loop.
 480
 481     break;
 482   }
 483 }
 484
 485 void SyncChannel::WaitForReplyWithNestedMessageLoop(SyncContext* context) {
 486   base::WaitableEventWatcher send_done_watcher;
 487
 488   ReceivedSyncMsgQueue* sync_msg_queue = context->received_sync_msgs();
 489   DCHECK(sync_msg_queue != NULL);
 490
 491   base::WaitableEventWatcher* old_send_done_event_watcher =
 492       sync_msg_queue->top_send_done_watcher();
 493
 494   base::WaitableEventWatcher::Delegate* old_delegate = NULL;
 495   base::WaitableEvent* old_event = NULL;
 496
 497   // Maintain a local global stack of send done delegates to ensure that
 498   // nested sync calls complete in the correct sequence, i.e. the
 499   // outermost call completes first, etc.
 500   if (old_send_done_event_watcher) {
 501     old_delegate = old_send_done_event_watcher->delegate();
 502     old_event = old_send_done_event_watcher->GetWatchedEvent();
 503     old_send_done_event_watcher->StopWatching();
 504   }
 505
 506   sync_msg_queue->set_top_send_done_watcher(&send_done_watcher);
 507
 508   send_done_watcher.StartWatching(context->GetSendDoneEvent(), context);
 509
 510   {
 511     MessageLoop::ScopedNestableTaskAllower allow(MessageLoop::current());
 512     MessageLoop::current()->Run();
 513   }
 514
 515   sync_msg_queue->set_top_send_done_watcher(old_send_done_event_watcher);
 516   if (old_send_done_event_watcher && old_event) {
 517     old_send_done_event_watcher->StartWatching(old_event, old_delegate);
 518   }
 519 }
 520
 521 void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) {
 522   DCHECK(event == sync_context()->GetDispatchEvent());
 523   // The call to DispatchMessages might delete this object, so reregister
 524   // the object watcher first.
 525   event->Reset();
 526   dispatch_watcher_.StartWatching(event, this);
 527   sync_context()->DispatchMessages();
 528 }
 529
 530 void SyncChannel::StartWatching() {
 531   // Ideally we only want to watch this object when running a nested message
 532   // loop.  However, we don't know when it exits if there's another nested
 533   // message loop running under it or not, so we wouldn't know whether to
 534   // stop or keep watching.  So we always watch it, and create the event as
 535   // manual reset since the object watcher might otherwise reset the event
 536   // when we're doing a WaitMany.
 537   dispatch_watcher_.StartWatching(sync_context()->GetDispatchEvent(), this);
 538 }
 539
 540 }  // namespace IPC