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