ipc/ipc_sync_message_filter.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_message_filter.h"
   6
   7 #include "base/bind.h"
   8 #include "base/location.h"
   9 #include "base/logging.h"
  10 #include "base/single_thread_task_runner.h"
  11 #include "base/synchronization/waitable_event.h"
  12 #include "base/thread_task_runner_handle.h"
  13 #include "ipc/ipc_channel.h"
  14 #include "ipc/ipc_sync_message.h"
  15
  16 namespace IPC {
  17
  18 bool SyncMessageFilter::Send(Message* message) {
  19   if (!message->is_sync()) {
  20     {
  21       base::AutoLock auto_lock(lock_);
  22       if (sender_ && is_channel_send_thread_safe_) {
  23         sender_->Send(message);
  24         return true;
  25       } else if (!io_task_runner_.get()) {
  26         pending_messages_.push_back(message);
  27         return true;
  28       }
  29     }
  30     io_task_runner_->PostTask(
  31         FROM_HERE,
  32         base::Bind(&SyncMessageFilter::SendOnIOThread, this, message));
  33     return true;
  34   }
  35
  36   base::WaitableEvent done_event(true, false);
  37   PendingSyncMsg pending_message(
  38       SyncMessage::GetMessageId(*message),
  39       static_cast<SyncMessage*>(message)->GetReplyDeserializer(),
  40       &done_event);
  41
  42   {
  43     base::AutoLock auto_lock(lock_);
  44     // Can't use this class on the main thread or else it can lead to deadlocks.
  45     // Also by definition, can't use this on IO thread since we're blocking it.
  46     if (base::ThreadTaskRunnerHandle::IsSet()) {
  47       DCHECK(base::ThreadTaskRunnerHandle::Get() != listener_task_runner_);
  48       DCHECK(base::ThreadTaskRunnerHandle::Get() != io_task_runner_);
  49     }
  50     pending_sync_messages_.insert(&pending_message);
  51
  52     if (io_task_runner_.get()) {
  53       io_task_runner_->PostTask(
  54           FROM_HERE,
  55           base::Bind(&SyncMessageFilter::SendOnIOThread, this, message));
  56     } else {
  57       pending_messages_.push_back(message);
  58     }
  59   }
  60
  61   base::WaitableEvent* events[2] = { shutdown_event_, &done_event };
  62   base::WaitableEvent::WaitMany(events, 2);
  63
  64   {
  65     base::AutoLock auto_lock(lock_);
  66     delete pending_message.deserializer;
  67     pending_sync_messages_.erase(&pending_message);
  68   }
  69
  70   return pending_message.send_result;
  71 }
  72
  73 void SyncMessageFilter::OnFilterAdded(Sender* sender) {
  74   std::vector<Message*> pending_messages;
  75   {
  76     base::AutoLock auto_lock(lock_);
  77     sender_ = sender;
  78     io_task_runner_ = base::ThreadTaskRunnerHandle::Get();
  79     pending_messages_.release(&pending_messages);
  80   }
  81   for (auto* msg : pending_messages)
  82     SendOnIOThread(msg);
  83 }
  84
  85 void SyncMessageFilter::OnChannelError() {
  86   base::AutoLock auto_lock(lock_);
  87   sender_ = NULL;
  88   SignalAllEvents();
  89 }
  90
  91 void SyncMessageFilter::OnChannelClosing() {
  92   base::AutoLock auto_lock(lock_);
  93   sender_ = NULL;
  94   SignalAllEvents();
  95 }
  96
  97 bool SyncMessageFilter::OnMessageReceived(const Message& message) {
  98   base::AutoLock auto_lock(lock_);
  99   for (PendingSyncMessages::iterator iter = pending_sync_messages_.begin();
 100        iter != pending_sync_messages_.end(); ++iter) {
 101     if (SyncMessage::IsMessageReplyTo(message, (*iter)->id)) {
 102       if (!message.is_reply_error()) {
 103         (*iter)->send_result =
 104             (*iter)->deserializer->SerializeOutputParameters(message);
 105       }
 106       (*iter)->done_event->Signal();
 107       return true;
 108     }
 109   }
 110
 111   return false;
 112 }
 113
 114 SyncMessageFilter::SyncMessageFilter(base::WaitableEvent* shutdown_event,
 115                                      bool is_channel_send_thread_safe)
 116     : sender_(NULL),
 117       is_channel_send_thread_safe_(is_channel_send_thread_safe),
 118       listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
 119       shutdown_event_(shutdown_event) {
 120 }
 121
 122 SyncMessageFilter::~SyncMessageFilter() {
 123 }
 124
 125 void SyncMessageFilter::SendOnIOThread(Message* message) {
 126   if (sender_) {
 127     sender_->Send(message);
 128     return;
 129   }
 130
 131   if (message->is_sync()) {
 132     // We don't know which thread sent it, but it doesn't matter, just signal
 133     // them all.
 134     base::AutoLock auto_lock(lock_);
 135     SignalAllEvents();
 136   }
 137
 138   delete message;
 139 }
 140
 141 void SyncMessageFilter::SignalAllEvents() {
 142   lock_.AssertAcquired();
 143   for (PendingSyncMessages::iterator iter = pending_sync_messages_.begin();
 144        iter != pending_sync_messages_.end(); ++iter) {
 145     (*iter)->done_event->Signal();
 146   }
 147 }
 148
 149 }  // namespace IPC